Repository: AgnostiqHQ/covalent
Pull request #1527 (2023-02-15 19:17:09)

Re-introduction: Implement dispatch cancellation (#1471)
This reverts commit a7a8987d459b8fe8d5597ef930236cce499e6264.

<!--
⚠️ Your pull request title should be short detailed and understandable for all.
⚠️ Increment the version number in the /VERSION file. If this is a bugfix, increment the patch version (the rightmost number), for example 0.18.2 becomes 0.18.3. If it's a new feature, increment the minor version (the middle number), for example 0.18.2 becomes 0.19.0.
⚠️ Add a note to /CHANGELOG.md with your version number and the date, summarizing the changes.
⚠️ If your pull request fixes an open issue, please link to the issue.
⚠️ Rebase the latest changes from the develop branch. Assuming origin points to https://github.com/AgnostiqHQ/covalent, you should run git rebase origin/develop.
-->

- [ ] I have added the tests to cover my changes.
- [ ] I have updated the documentation and CHANGELOG accordingly.
- [ ] I have read the CONTRIBUTING document.


codecov[bot]: # [Codecov](https://codecov.io/gh/AgnostiqHQ/covalent/pull/1527?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) Report
> Merging [#1527](https://codecov.io/gh/AgnostiqHQ/covalent/pull/1527?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (c183df4) into [develop](https://codecov.io/gh/AgnostiqHQ/covalent/commit/2a66ee509c9b0b2d43cd9df8e18e7475e03c9088?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (2a66ee5) will **increase** coverage by `0.43%`.
> The diff coverage is `98.29%`.

<details><summary>Additional details and impacted files</summary>


[![Impacted file tree graph](https://codecov.io/gh/AgnostiqHQ/covalent/pull/1527/graphs/tree.svg?width=650&height=150&src=pr&token=YGHCB3DE4P&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ)](https://codecov.io/gh/AgnostiqHQ/covalent/pull/1527?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ)

```diff
@@             Coverage Diff             @@
##           develop    #1527      +/-   ##
===========================================
+ Coverage    85.05%   85.48%   +0.43%     
===========================================
  Files          146      149       +3     
  Lines         5940     6257     +317     
  Branches       145      145              
===========================================
+ Hits          5052     5349     +297     
- Misses         786      806      +20     
  Partials       102      102              
```

| Flag | Coverage Δ | |
|---|---|---|
| Dispatcher | `88.76% <98.33%> (+0.19%)` | :arrow_up: |
| Functional_Tests | `57.67% <43.40%> (-2.00%)` | :arrow_down: |
| SDK | `83.77% <97.27%> (+0.61%)` | :arrow_up: |
| UI_Frontend | `72.13% <ø> (ø)` | |

</details>

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index fa617609..0f9bb314 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [UNRELEASED]
 
+### Added
+
+- Added feature to support cancelling workflow dispatches
+- Updating/adding new tests to improve code coverage
+
 ### Fixed
 
 - Redispatch bug involving copying reusable nodes from old transport graph to new transport graph.
diff --git a/covalent/_results_manager/results_manager.py b/covalent/_results_manager/results_manager.py
index 7b23641a..5ec53125 100644
--- a/covalent/_results_manager/results_manager.py
+++ b/covalent/_results_manager/results_manager.py
@@ -193,23 +193,31 @@ def sync(
         )
 
 
-def cancel(
-    dispatch_id: str,
-    dispatcher: str = get_config("dispatcher.address") + ":" + str(get_config("dispatcher.port")),
-) -> str:
+def cancel(dispatch_id: str, task_ids: List[int] = None, dispatcher_addr: str = None) -> str:
     """
     Cancel a running dispatch.
 
     Args:
         dispatch_id: The dispatch id of the dispatch to be cancelled.
-        dispatcher: The address to the dispatcher in the form of hostname:port, e.g. "localhost:8080".
+        dispatcher_addr: Dispatcher server address, if None then defaults to the address set in Covalent's config.
 
     Returns:
         None
     """
 
-    url = "http://" + dispatcher + "/api/cancel"
+    if dispatcher_addr is None:
+        dispatcher_addr = (
+            get_config("dispatcher.address") + ":" + str(get_config("dispatcher.port"))
+        )
+
+    if task_ids is None:
+        task_ids = []
+
+    url = f"http://{dispatcher_addr}/api/cancel"
+
+    if isinstance(task_ids, int):
+        task_ids = [task_ids]
 
-    r = requests.post(url, data=dispatch_id.encode("utf-8"))
+    r = requests.post(url, json={"dispatch_id": dispatch_id, "task_ids": task_ids})
     r.raise_for_status()
     return r.content.decode("utf-8").strip().replace('"', "")
diff --git a/covalent/_shared_files/__init__.py b/covalent/_shared_files/__init__.py
index 520a1dff..52004d44 100644
--- a/covalent/_shared_files/__init__.py
+++ b/covalent/_shared_files/__init__.py
@@ -21,4 +21,4 @@
 """Import shared utils."""
 
 from .defaults import DefaultMetadataValues
-from .exceptions import TaskRuntimeError
+from .exceptions import TaskCancelledError, TaskRuntimeError
diff --git a/covalent/_shared_files/exceptions.py b/covalent/_shared_files/exceptions.py
index 5b6ec6fa..3b57c52c 100644
--- a/covalent/_shared_files/exceptions.py
+++ b/covalent/_shared_files/exceptions.py
@@ -25,3 +25,7 @@ class MissingLatticeRecordError(Exception):
 
 class TaskRuntimeError(Exception):
     pass
+
+
+class TaskCancelledError(Exception):
+    pass
diff --git a/covalent/_workflow/transport.py b/covalent/_workflow/transport.py
index f6888754..9892f1b3 100644
--- a/covalent/_workflow/transport.py
+++ b/covalent/_workflow/transport.py
@@ -383,9 +383,8 @@ class _TransportGraph:
             value: The value from the node stored at the value key.
 
         Raises:
-            KeyError: If the value key is not found.
+            KeyError: If the value key or node key is not found.
         """
-
         return self._graph.nodes[node_key][value_key]
 
     def set_node_value(self, node_key: int, value_key: int, value: Any) -> None:
diff --git a/covalent/executor/base.py b/covalent/executor/base.py
index 615dd914..ff46a1a0 100644
--- a/covalent/executor/base.py
+++ b/covalent/executor/base.py
@@ -25,22 +25,39 @@ Class that defines the base executor template.
 import asyncio
 import copy
 import io
+import json
 import os
+import queue
 from abc import ABC, abstractmethod
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
-from typing import Any, Callable, ContextManager, Dict, Iterable, List, Tuple
+from typing import (
+    Any,
+    Callable,
+    ContextManager,
+    Dict,
+    Iterable,
+    List,
+    Literal,
+    Optional,
+    Tuple,
+    Union,
+)
 
 import aiofiles
 
+from covalent._shared_files.exceptions import TaskCancelledError
 from covalent._workflow.depscall import RESERVED_RETVAL_KEY__FILES
+from covalent.executor.utils import Signals
 
 from .._shared_files import TaskRuntimeError, logger
 from .._shared_files.context_managers import active_dispatch_info_manager
-from .._shared_files.util_classes import DispatchInfo
+from .._shared_files.util_classes import RESULT_STATUS, DispatchInfo
 from .._workflow.transport import TransportableObject
 
 app_log = logger.app_log
 log_stack_info = logger.log_stack_info
+TypeJSON = Union[str, int, float, bool, None, Dict[str, Any], List[Any]]
 
 
 def wrapper_fn(
@@ -202,6 +219,89 @@ class BaseExecutor(_AbstractBaseExecutor):
     ) -> None:
         super().__init__(*args, **kwargs)
 
+    def _init_runtime(
+        self,
+        loop: Optional[asyncio.AbstractEventLoop] = None,
+        cancel_pool: Optional[ThreadPoolExecutor] = None,
+    ) -> None:
+        """
+        Create the required queues for cancel task messages to be shared back and forth
+
+        Arg(s)
+            loop: Asyncio event loop to create tasks on
+            cancel_pool: A ThreadPoolExecutor object to submit tasks to
+
+        Return(s)
+            None
+        """
+        self._send_queue = asyncio.Queue()
+        self._recv_queue = queue.Queue()
+        self._loop = loop
+        self._cancel_pool = cancel_pool
+
+    def _notify(self, action: Signals, body: Any = None) -> None:
+        """
+        Notifies a waiting thread with the necessary action to take along with the arguments passed in as body
+
+        Arg(s)
+            action: One of three possible actions that a waiting thread must take -> Signals.GET, Signals.PUT, Signals.EXIT
+            body: Respective arguments for each action
+
+        Return(s)
+            None
+        """
+        self._loop.call_soon_threadsafe(self._send_queue.put_nowait, (action, body))
+
+    def _notify_sync(self, action: Signals, body: Any = None) -> Any:
+        """
+        Blocking version of the _notify method
+
+        Arg(s)
+            action: One of three possible actions that a waiting thread must take -> Signals.GET, Signals.PUT, Signals.EXIT
+            body: Respective arguments for each action
+        """
+        self._notify(action, body)
+        return self._wait_for_response()
+
+    def _wait_for_response(self, timeout: int = 5) -> Any:
+        """
+        Wait for response from a thread depending on the action/body parameters sent
+
+        Arg(s):
+            timeout: Number of seconds to wait for the result to become available in the queue. Defaults to five seconds
+
+        Return(s)
+            body: Response to the corresponding action
+        """
+        status, body = self._recv_queue.get(timeout=timeout)
+        if status is None:
+            raise RuntimeError("Error waiting for response")
+        return body
+
+    def get_cancel_requested(self) -> bool:
+        """
+        Check if the task was requested to be cancelled by the user
+
+        Arg(s)
+            None
+
+        Return(s)
+            True/False whether task cancellation was requested
+        """
+        return self._notify_sync(Signals.GET, "cancel_requested")
+
+    def set_job_handle(self, handle: TypeJSON) -> Any:
+        """
+        Save the job_id/handle returned by the backend executing the task
+
+        Arg(s)
+            handle: Any JSONable type to identifying the task being executed by the backend
+
+        Return(s)
+            Response from saving the job handle to database
+        """
+        return self._notify_sync(Signals.PUT, ("job_handle", json.dumps(handle)))
+
     def write_streams_to_file(
         self,
         stream_strings: Iterable[str],
@@ -293,17 +393,18 @@ class BaseExecutor(_AbstractBaseExecutor):
             "results_dir": results_dir,
         }
 
-        self.setup(task_metadata=task_metadata)
-
         try:
+            self.setup(task_metadata=task_metadata)
             result = self.run(function, args, kwargs, task_metadata)
-            exception_raised = False
+            job_status = RESULT_STATUS.COMPLETED
         except TaskRuntimeError as err:
-            app_log.error(f"TaskRuntimeError: {err}")
-            exception_raised = True
+            job_status = RESULT_STATUS.FAILED
+            result = None
+        except TaskCancelledError as err:
+            job_status = RESULT_STATUS.CANCELLED
             result = None
-
         finally:
+            self._notify(Signals.EXIT)
             self.teardown(task_metadata=task_metadata)
 
         self.write_streams_to_file(
@@ -313,12 +414,7 @@ class BaseExecutor(_AbstractBaseExecutor):
             results_dir,
         )
 
-        return (
-            result,
-            self._task_stdout.getvalue(),
-            self._task_stderr.getvalue(),
-            exception_raised,
-        )
+        return (result, self._task_stdout.getvalue(), self._task_stderr.getvalue(), job_status)
 
     def setup(self, task_metadata: Dict) -> Any:
         """Placeholder to run any executor specific tasks"""
@@ -341,6 +437,37 @@ class BaseExecutor(_AbstractBaseExecutor):
 
         raise NotImplementedError
 
+    def cancel(self, task_metadata: Dict, job_handle: Any) -> Literal[False]:
+        """
+        Method to cancel the job identified uniquely by the `job_handle` (base class)
+
+        Arg(s)
+            task_metadata: Metadata of the task to be cancelled
+            job_handle: Unique ID of the job assigned by the backend
+
+        Return(s)
+            False by default
+        """
+        app_log.debug(f"Cancel not implemented for executor {type(self)}")
+        return False
+
+    async def _cancel(self, task_metadata: Dict, job_handle: Any) -> Any:
+        """
+        Cancel the task in a non-blocking manner
+
+        Arg(s)
+            task_metadata: Metadata of the task to be cancelled
+            job_handle: Unique ID of the job assigned by the backend
+
+        Return(s)
+           Result of the task cancellation
+        """
+        cancel_result = await self._loop.run_in_executor(
+            self._cancel_pool, self.cancel, task_metadata, job_handle
+        )
+        await self._loop.run_in_executor(self._cancel_pool, self.teardown, task_metadata)
+        return cancel_result
+
     def teardown(self, task_metadata: Dict) -> Any:
         """Placeholder to run nay executor specific cleanup/teardown actions"""
         pass
@@ -371,6 +498,91 @@ class AsyncBaseExecutor(_AbstractBaseExecutor):
     ) -> None:
         super().__init__(*args, **kwargs)
 
+    def _init_runtime(
+        self,
+        loop: Optional[asyncio.AbstractEventLoop] = None,
+        cancel_pool: Optional[ThreadPoolExecutor] = None,
+    ) -> None:
+        """
+        Initialize the send and receive queues for communication between dispatcher and the executor
+
+        Arg(s)
+            loop: Asyncio event loop
+            cancel_pool: Instance of a threadpool executor class
+
+        Return(s)
+            None
+        """
+        self._send_queue = asyncio.Queue()
+        self._recv_queue = asyncio.Queue()
+
+    def _notify(self, action: Signals, body: Any = None) -> None:
+        """
+        Notify the listener with the corresponding signal (async)
+
+        Arg(s)
+            action: Signal to the listener to trigger the corresponding action
+            body: Message to be sent to the listener
+
+        Return(s)
+            None
+        """
+        self._send_queue.put_nowait((action, body))
+
+    async def _notify_sync(self, action: Signals, body: Any = None) -> Any:
+        """
+        Blocking call to the `_notify` method to wait for response
+
+        Arg(s)
+            action: Signal to the listener to trigger the corresponding action
+            body: Message to be sent to the listener
+
+        Return(s)
+            Response from the listener
+        """
+        self._notify(action, body)
+        return await self._wait_for_response()
+
+    async def _wait_for_response(self, timeout: int = 5) -> Any:
+        """
+        Block the thread until a response is recevied
+
+        Arg(s)
+            timeout: Number of seconds to wait until timing out
+
+        Return(s)
+            Response from the listener
+        """
+        aw = self._recv_queue.get()
+        status, body = await asyncio.wait_for(aw, timeout=timeout)
+        if status is False:
+            raise RuntimeError("Error waiting for response")
+        return body
+
+    async def get_cancel_requested(self) -> Any:
+        """
+        Get if the task was requested to be canceled
+
+        Arg(s)
+            None
+
+        Return(s)
+            Whether the task has been requested to be cancelled
+        """
+        return await self._notify_sync(Signals.GET, "cancel_requested")
+
+    async def set_job_handle(self, handle: TypeJSON) -> Any:
+        """
+        Save the job handle to database
+
+        Arg(s)
+            handle: JSONable type identifying the job being executed by the backend
+
+        Return(s)
+            Response from the listener that handles inserting the job handle to database
+        """
+        return await self._notify_sync(Signals.PUT, ("job_handle", json.dumps(handle)))
+
     async def write_streams_to_file(
         self,
         stream_strings: Iterable[str],
@@ -440,16 +652,18 @@ class AsyncBaseExecutor(_AbstractBaseExecutor):
             "results_dir": results_dir,
         }
 
-        await self.setup(task_metadata=task_metadata)
-
         try:
+            await self.setup(task_metadata=task_metadata)
             result = await self.run(function, args, kwargs, task_metadata)
-            exception_raised = False
+            job_status = RESULT_STATUS.COMPLETED
+        except TaskCancelledError as err:
+            job_status = RESULT_STATUS.CANCELLED
+            result = None
         except TaskRuntimeError as err:
-            exception_raised = True
+            job_status = RESULT_STATUS.FAILED
             result = None
-
         finally:
+            self._notify(Signals.EXIT)
             await self.teardown(task_metadata=task_metadata)
 
         await self.write_streams_to_file(
@@ -459,12 +673,7 @@ class AsyncBaseExecutor(_AbstractBaseExecutor):
             results_dir,
         )
 
-        return (
-            result,
-            self._task_stdout.getvalue(),
-            self._task_stderr.getvalue(),
-            exception_raised,
-        )
+        return (result, self._task_stdout.getvalue(), self._task_stderr.getvalue(), job_status)
 
     async def setup(self, task_metadata: Dict):
         """Executor specific setup method"""
@@ -490,3 +699,32 @@ class AsyncBaseExecutor(_AbstractBaseExecutor):
         """
 
         raise NotImplementedError
+
+    async def cancel(self, task_metadata: Dict, job_handle: Any) -> Literal[False]:
+        """
+        Executor specific task cancellation method
+
+        Arg(s)
+            task_metadata: Metadata associated with the task to be cancelled
+            job_handle: Unique ID assigned to the job by the backend
+
+        Return(s)
+            False by default
+        """
+        app_log.debug(f"Cancel not implemented for executor {type(self)}")
+        return False
+
+    async def _cancel(self, task_metadata: Dict, job_handle: Any) -> Literal[False]:
+        """
+        Cancel the task in a non-blocking manner and teardown the infrastructure
+
+        Arg(s)
+            task_metadata: Metadata associated with the task to be cancelled
+            job_handle: Unique ID assigned to the job by the backend
+
+        Return(s)
+            Result from cancelling the task
+        """
+        cancel_result = await self.cancel(task_metadata, job_handle)
+        await self.teardown(task_metadata)
+        return cancel_result
diff --git a/covalent/executor/executor_plugins/dask.py b/covalent/executor/executor_plugins/dask.py
index 96b9b8bc..c1ae4c1e 100644
--- a/covalent/executor/executor_plugins/dask.py
+++ b/covalent/executor/executor_plugins/dask.py
@@ -26,14 +26,15 @@ This is a plugin executor module; it is loaded if found and properly structured.
 """
 
 import os
-from typing import Callable, Dict, List
+from typing import Callable, Dict, List, Literal
 
-from dask.distributed import Client
+from dask.distributed import CancelledError, Client, Future
 
 from covalent._shared_files import TaskRuntimeError, logger
 
 # Relative imports are not allowed in executor plugins
 from covalent._shared_files.config import get_config
+from covalent._shared_files.exceptions import TaskCancelledError
 from covalent._shared_files.utils import _address_client_mapper
 from covalent.executor.base import AsyncBaseExecutor
 from covalent.executor.utils.wrappers import io_wrapper as dask_wrapper
@@ -88,6 +89,10 @@ class DaskExecutor(AsyncBaseExecutor):
     async def run(self, function: Callable, args: List, kwargs: Dict, task_metadata: Dict):
         """Submit the function and inputs to the dask cluster"""
 
+        if await self.get_cancel_requested():
+            app_log.debug("Task has cancelled")
+            raise TaskCancelledError
+
         node_id = task_metadata["node_id"]
 
         dask_client = _address_client_mapper.get(self.scheduler_address)
@@ -95,13 +100,16 @@ class DaskExecutor(AsyncBaseExecutor):
         if not dask_client:
             dask_client = Client(address=self.scheduler_address, asynchronous=True)
             _address_client_mapper[self.scheduler_address] = dask_client
-
             await dask_client
 
         future = dask_client.submit(dask_wrapper, function, args, kwargs)
-        app_log.debug(f"Submitted task {node_id} to dask")
+        await self.set_job_handle(future.key)
+        app_log.debug(f"Submitted task {node_id} to dask with key {future.key}")
 
-        result, worker_stdout, worker_stderr, tb = await future
+        try:
+            result, worker_stdout, worker_stderr, tb = await future
+        except CancelledError:
+            raise TaskCancelledError()
 
         print(worker_stdout, end="", file=self.task_stdout)
         print(worker_stderr, end="", file=self.task_stderr)
@@ -112,3 +120,26 @@ class DaskExecutor(AsyncBaseExecutor):
 
         # FIX: need to get stdout and stderr from dask worker and print them
         return result
+
+    async def cancel(self, task_metadata: Dict, job_handle) -> Literal[True]:
+        """
+        Cancel the task being executed by the dask executor currently
+
+        Arg(s)
+            task_metadata: Metadata associated with the task
+            job_handle: Key assigned to the job by Dask
+
+        Return(s)
+            True by default
+        """
+        dask_client = _address_client_mapper.get(self.scheduler_address)
+
+        if not dask_client:
+            dask_client = Client(address=self.scheduler_address, asynchronous=True)
+            _address_client_mapper[self.scheduler_address] = dask_client
+            await dask_client
+
+        fut: Future = Future(key=job_handle, client=dask_client)
+        await fut.cancel()
+        app_log.debug(f"Cancelled future with key {job_handle}")
+        return True
diff --git a/covalent/executor/executor_plugins/local.py b/covalent/executor/executor_plugins/local.py
index 971f32e2..f5e2497e 100644
--- a/covalent/executor/executor_plugins/local.py
+++ b/covalent/executor/executor_plugins/local.py
@@ -27,10 +27,10 @@ This is a plugin executor module; it is loaded if found and properly structured.
 
 import os
 from concurrent.futures import ProcessPoolExecutor
-from typing import Callable, Dict, List
+from typing import Any, Callable, Dict, List
 
 # Relative imports are not allowed in executor plugins
-from covalent._shared_files import TaskRuntimeError, logger
+from covalent._shared_files import TaskCancelledError, TaskRuntimeError, logger
 from covalent.executor import BaseExecutor
 
 # Store the wrapper function in an external module to avoid module
@@ -59,9 +59,27 @@ class LocalExecutor(BaseExecutor):
     Local executor class that directly invokes the input function.
     """
 
-    def run(self, function: Callable, args: List, kwargs: Dict, task_metadata: Dict):
+    def run(self, function: Callable, args: List, kwargs: Dict, task_metadata: Dict) -> Any:
+        """
+        Execute the function locally
+
+        Arg(s)
+            function: Function to be executed
+            args: Arguments passed to the function
+            kwargs: Keyword arguments passed to the function
+            task_metadata: Metadata of the task to be executed
+
+        Return(s)
+            Task output
+        """
         app_log.debug(f"Running function {function} locally")
 
+        self.set_job_handle(42)
+
+        if self.get_cancel_requested():
+            app_log.debug("Task has been cancelled don't proceed")
+            raise TaskCancelledError
+
         # Run the target function in a separate process
         fut = proc_pool.submit(io_wrapper, function, args, kwargs)
 
diff --git a/covalent/executor/utils/__init__.py b/covalent/executor/utils/__init__.py
index 523f7762..45949792 100644
--- a/covalent/executor/utils/__init__.py
+++ b/covalent/executor/utils/__init__.py
@@ -17,3 +17,5 @@
 # FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
 #
 # Relief from the License may be granted by purchasing a commercial license.
+
+from .wrappers import Signals
diff --git a/covalent/executor/utils/wrappers.py b/covalent/executor/utils/wrappers.py
index f140e44b..c39a3802 100644
--- a/covalent/executor/utils/wrappers.py
+++ b/covalent/executor/utils/wrappers.py
@@ -25,9 +25,20 @@ Helper functions for the local executor
 import io
 import traceback
 from contextlib import redirect_stderr, redirect_stdout
+from enum import Enum
 from typing import Any, Callable, Dict, List, Tuple
 
 
+class Signals(Enum):
+    """
+    Signals to enable communication between the executors and Covalent dispatcher
+    """
+
+    GET = 0
+    PUT = 1
+    EXIT = 2
+
+
 def io_wrapper(fn: Callable, args: List, kwargs: Dict) -> Tuple[Any, str, str, str]:
     """Wrapper function to execute the given function in a separate
     process and capture stdout and stderr"""
diff --git a/covalent_dispatcher/_core/__init__.py b/covalent_dispatcher/_core/__init__.py
index 59426d08..eb298676 100644
--- a/covalent_dispatcher/_core/__init__.py
+++ b/covalent_dispatcher/_core/__init__.py
@@ -19,4 +19,4 @@
 # Relief from the License may be granted by purchasing a commercial license.
 
 from .data_manager import make_derived_dispatch, make_dispatch
-from .dispatcher import cancel_workflow, run_dispatch
+from .dispatcher import cancel_dispatch, run_dispatch
diff --git a/covalent_dispatcher/_core/data_manager.py b/covalent_dispatcher/_core/data_manager.py
index d2012945..9da0030d 100644
--- a/covalent_dispatcher/_core/data_manager.py
+++ b/covalent_dispatcher/_core/data_manager.py
@@ -46,7 +46,7 @@ _dispatch_status_queues = {}
 
 
 def generate_node_result(
-    node_id,
+    node_id: int,
     start_time=None,
     end_time=None,
     status=None,
@@ -57,6 +57,24 @@ def generate_node_result(
     sub_dispatch_id=None,
     sublattice_result=None,
 ):
+    """
+    Helper routine to prepare the node result
+
+    Arg(s)
+        node_id: ID of the node in the trasport graph
+        start_time: Start time of the node
+        end_time: Time at which the node finished executing
+        status: Status of the node's execution
+        output: Output of the node
+        error: Error from the node
+        stdout: STDOUT of a node
+        stderr: STDERR generated during node execution
+        sub_dispatch_id: Dispatch ID of the sublattice
+        sublattice_result: Result of the sublattice
+
+    Return(s)
+        Dictionary of the inputs
+    """
     return {
         "node_id": node_id,
         "start_time": start_time,
@@ -72,12 +90,22 @@ def generate_node_result(
 
 
 # Domain: result
-async def update_node_result(result_object, node_result):
+async def update_node_result(result_object, node_result) -> None:
+    """
+    Updates the result object with the current node_result
+
+    Arg(s)
+        result_object: Result object the current dispatch
+        node_result: Result of the node to be updated in the result object
+
+    Return(s)
+        None
+    """
     app_log.warning("Updating node result (run_planned_workflow).")
     try:
         update._node(result_object, **node_result)
     except Exception as ex:
-        app_log.exception("Error persisting node update: {ex}")
+        app_log.exception(f"Error persisting node update: {ex}")
         node_result["status"] = Result.FAILED
     finally:
         node_id = node_result["node_id"]
@@ -251,4 +279,4 @@ async def _update_parent_electron(result_object: Result):
 
 def upsert_lattice_data(dispatch_id: str):
     result_object = get_result_object(dispatch_id)
-    upsert._lattice_data(result_object)
+    upsert.lattice_data(result_object)
diff --git a/covalent_dispatcher/_core/data_modules/__init__.py b/covalent_dispatcher/_core/data_modules/__init__.py
new file mode 100644
index 00000000..9d1b0552
--- /dev/null
+++ b/covalent_dispatcher/_core/data_modules/__init__.py
@@ -0,0 +1,19 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
diff --git a/covalent_dispatcher/_core/data_modules/job_manager.py b/covalent_dispatcher/_core/data_modules/job_manager.py
new file mode 100644
index 00000000..1500e1aa
--- /dev/null
+++ b/covalent_dispatcher/_core/data_modules/job_manager.py
@@ -0,0 +1,117 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+# """Interface to the Jobs table"""
+
+from typing import Any, List
+
+from ..._db.jobdb import get_job_records, to_job_ids, update_job_records
+
+
+def _set_cancel_requested(job_ids: List[int]) -> None:
+    """
+    Update the job record with `cancel_requested`= True (private to module)
+
+    Arg(s)
+        job_ids: List of all job ids that have been requested to be cancelled
+
+    Return(s)
+        None
+    """
+    records = [{"job_id": job_id, "cancel_requested": True} for job_id in job_ids]
+    update_job_records(records)
+
+
+async def set_cancel_requested(dispatch_id: str, task_ids: List[int]):
+    """
+    Set all the task with `task_ids` to be cancelled
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the workflow
+        task_ids: List of task ids to be cancelled
+
+    Return(s)
+        None
+    """
+    job_ids = to_job_ids(dispatch_id, task_ids)
+    _set_cancel_requested(job_ids)
+
+
+async def get_jobs_metadata(dispatch_id: str, task_ids: List[int]) -> Any:
+    """
+    Retrive all job records with task_ids for the given dispatch
+
+    Arg(s)
+        dispatch_id: Dispatch ID
+        task_ids: List of task ids
+
+    Return(s)
+        Dictionary of job metdata associated with each task
+    """
+    job_ids = to_job_ids(dispatch_id, task_ids)
+    return get_job_records(job_ids)
+
+
+async def _set_job_metadata(dispatch_id: str, task_id: int, **kwargs) -> None:
+    """
+    Store any metadata associated with a job to the database
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the the lattice
+        task_id: ID of the task in the lattice
+        kwargs: Keyword arguments
+
+    Return(s)
+        None
+    """
+    job_id = to_job_ids(dispatch_id, [task_id])[0]
+    update_kwargs = kwargs
+    update_kwargs["job_id"] = job_id
+    update_job_records([update_kwargs])
+
+
+async def set_job_handle(dispatch_id: str, task_id: int, job_handle: str) -> None:
+    """
+    Set the job handle of the task to the database
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+        task_id: ID of the task in the lattice
+        job_handle: Job handle returned by the backend corresponding to the task (JSONable)
+
+    Return(s)
+        None
+    """
+    await _set_job_metadata(dispatch_id, task_id, job_handle=job_handle)
+
+
+async def set_cancel_result(dispatch_id: str, task_id: int, cancel_status: bool) -> None:
+    """
+    Update the cancel status of the job in the database if task cancellation is requested
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+        task_id: ID of the task in the lattice
+        cancel_status: True/False indicating whether the task is to be cancelled
+
+    Return(s)
+        None
+    """
+    await _set_job_metadata(dispatch_id, task_id, cancel_successful=cancel_status)
diff --git a/covalent_dispatcher/_core/dispatcher.py b/covalent_dispatcher/_core/dispatcher.py
index 00adeb10..b731c5ad 100644
--- a/covalent_dispatcher/_core/dispatcher.py
+++ b/covalent_dispatcher/_core/dispatcher.py
@@ -25,7 +25,7 @@ Defines the core functionality of the dispatcher
 import asyncio
 import traceback
 from datetime import datetime, timezone
-from typing import Dict, Tuple
+from typing import Dict, List, Tuple
 
 from covalent._results_manager import Result
 from covalent._shared_files import logger
@@ -34,6 +34,7 @@ from covalent_ui import result_webhook
 
 from . import data_manager as datasvc
 from . import runner
+from .data_modules.job_manager import set_cancel_requested
 
 app_log = logger.app_log
 log_stack_info = logger.log_stack_info
@@ -76,12 +77,23 @@ def _get_abstract_task_inputs(node_id: int, node_name: str, result_object: Resul
 
 # Domain: dispatcher
 async def _handle_completed_node(result_object, node_id, pending_parents):
+    """
+    Process the completed node in the transport graph
+
+    Arg(s)
+        result_object: Result object associated with the workflow
+        node_id: ID of the node in the transport graph
+        pending_parents: Parents of this node yet to be executed
+
+    Return(s)
+        List of nodes ready to be executed
+    """
     g = result_object.lattice.transport_graph._graph
 
     ready_nodes = []
     app_log.debug(f"Node {node_id} completed")
     for child, edges in g.adj[node_id].items():
-        for edge in edges:
+        for _ in edges:
             pending_parents[child] -= 1
         if pending_parents[child] < 1:
             app_log.debug(f"Queuing node {child} for execution")
@@ -352,23 +364,45 @@ async def run_workflow(result_object: Result) -> Result:
 
 
 # Domain: dispatcher
-def cancel_workflow(dispatch_id: str) -> None:
+async def cancel_dispatch(dispatch_id: str, task_ids: List[int] = []) -> None:
     """
-    Cancels a dispatched workflow using publish subscribe mechanism
-    provided by Dask.
+    Cancel an entire dispatch or a specific set of tasks within it
 
-    Args:
-        dispatch_id: Dispatch id of the workflow to be cancelled
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+        task_ids: List of tasks from the lattice that are to be cancelled. Defaults to [] (entire lattice)
 
-    Returns:
+    Return(s)
         None
     """
+    if not dispatch_id:
+        return
 
-    # shared_var = Variable(dispatch_id)
-    # shared_var.set(str(Result.CANCELLED))
-    pass
+    tg = datasvc.get_result_object(dispatch_id=dispatch_id).lattice.transport_graph
+    if task_ids:
+        app_log.debug(f"Cancelling tasks {task_ids} in dispatch {dispatch_id}")
+    else:
+        task_ids = list(tg._graph.nodes)
+        app_log.debug(f"Cancelling dispatch {dispatch_id}")
+
+    await set_cancel_requested(dispatch_id, task_ids)
+    await runner.cancel_tasks(dispatch_id, task_ids)
+
+    # Recursively cancel running sublattice dispatches
+    sub_ids = list(map(lambda x: tg.get_node_value(x, "sub_dispatch_id"), task_ids))
+    for sub_dispatch_id in sub_ids:
+        await cancel_dispatch(sub_dispatch_id)
 
 
 def run_dispatch(dispatch_id: str) -> asyncio.Future:
+    """
+    Run the workflow and return immediately
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+
+    Return(s)
+        asyncio.Future
+    """
     result_object = datasvc.get_result_object(dispatch_id)
     return asyncio.create_task(run_workflow(result_object))
diff --git a/covalent_dispatcher/_core/execution.py b/covalent_dispatcher/_core/execution.py
index d498f1e6..20c3315d 100644
--- a/covalent_dispatcher/_core/execution.py
+++ b/covalent_dispatcher/_core/execution.py
@@ -22,38 +22,6 @@
 Defines the core functionality of the dispatcher
 """
 
-# Legacy imports
-# from .dispatcher import (
-#     _build_sublattice_graph,
-#     _dispatch_sync_sublattice,
-#     _get_abstract_task_inputs,
-#     _handle_cancelled_node,
-#     _handle_completed_node,
-#     _handle_failed_node,
-#     _initialize_deps_and_queue,
-#     _plan_workflow,
-#     _post_process,
-#     _postprocess_workflow,
-#     _run_planned_workflow,
-#     _submit_task,
-#     cancel_workflow,
-#     run_workflow,
-# )
-# from .result import (
-#     _update_node_result,
-#     generate_node_result,
-#     get_unique_id,
-#     initialize_result_object,
-# )
-# from .runner import (
-#     _gather_deps,
-#     _get_task_input_values,
-#     _get_task_inputs,
-#     _run_abstract_task,
-#     _run_task,
-#     _run_task_and_update,
-# )
-
 from covalent._results_manager import Result
 
 from . import dispatcher, runner
diff --git a/covalent_dispatcher/_core/runner.py b/covalent_dispatcher/_core/runner.py
index 4bbcecb3..6644e58f 100644
--- a/covalent_dispatcher/_core/runner.py
+++ b/covalent_dispatcher/_core/runner.py
@@ -22,11 +22,13 @@
 Defines the core functionality of the runner
 """
 
+import asyncio
 import json
 import traceback
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
 from functools import partial
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, List, Literal, Tuple, Union
 
 from covalent._results_manager import Result
 from covalent._shared_files import logger
@@ -43,14 +45,28 @@ from .._db import upsert
 from .._db.write_result_to_db import get_sublattice_electron_id
 from . import data_manager as datasvc
 from . import dispatcher
+from .data_modules.job_manager import get_jobs_metadata, set_cancel_result
+from .runner_modules import executor_proxy
 
 app_log = logger.app_log
 log_stack_info = logger.log_stack_info
 debug_mode = get_config("sdk.log_level") == "debug"
 
+_cancel_threadpool = ThreadPoolExecutor()
+
 
 # This is to be run out-of-process
 def _build_sublattice_graph(sub: Lattice, parent_metadata: Dict, *args, **kwargs):
+    """
+    Build the sublattice graph dynamically
+
+    Arg(s)
+        sub: Lattice associated with the sub-lattice
+        parent_metadata: Metadata of the parent lattice
+
+    Return(s)
+        JSON representation of the sublattice transport graph
+    """
     for k in sub.metadata.keys():
         if not sub.metadata[k]:
             sub.metadata[k] = parent_metadata[k]
@@ -116,12 +132,22 @@ async def _dispatch_sublattice(
     else:
         app_log.debug("Error building sublattice graph")
         stderr = res["stderr"]
-        raise RuntimeError("Error building sublattice graph: {stderr}")
+        raise RuntimeError(f"Error building sublattice graph: {stderr}")
 
 
 # Domain: runner
 # to be called by _run_abstract_task
 def _get_task_input_values(result_object: Result, abs_task_inputs: dict) -> dict:
+    """
+    Retrive the input values from the result_object for the task
+
+    Arg(s)
+        result_object: Result object of the workflow
+        abs_task_inputs: Task inputs dictionary
+
+    Return(s)
+        node_values: Dictionary of task inputs
+    """
     node_values = {}
     args = abs_task_inputs["args"]
     for node_id in args:
@@ -129,7 +155,7 @@ def _get_task_input_values(result_object: Result, abs_task_inputs: dict) -> dict
         node_values[node_id] = value
 
     kwargs = abs_task_inputs["kwargs"]
-    for key, node_id in kwargs.items():
+    for _, node_id in kwargs.items():
         value = result_object.lattice.transport_graph.get_node_value(node_id, "output")
         node_values[node_id] = value
 
@@ -150,6 +176,15 @@ async def _run_abstract_task(
     timestamp = datetime.now(timezone.utc)
 
     try:
+        cancel_req = await _get_cancel_requested(dispatch_id, node_id)
+        if cancel_req:
+            app_log.debug(f"Don't run cancelled task {dispatch_id}:{node_id}")
+            return datasvc.generate_node_result(
+                node_id=node_id,
+                start_time=timestamp,
+                end_time=timestamp,
+                status=Result.CANCELLED,
+            )
         serialized_callable = result_object.lattice.transport_graph.get_node_value(
             node_id, "function"
         )
@@ -239,6 +274,7 @@ async def _run_task(
         # the executor is determined during scheduling and provided in the execution metadata
         executor = _executor_manager.get_executor(short_name)
         executor.from_dict(object_dict)
+        executor._init_runtime(loop=asyncio.get_running_loop())
     except Exception as ex:
         tb = "".join(traceback.TracebackException.from_exception(ex).format())
         app_log.debug("Exception when trying to instantiate executor:")
@@ -288,7 +324,10 @@ async def _run_task(
                 results_dir=results_dir,
                 node_id=node_id,
             )
-            output, stdout, stderr, exception_raised = await executor._execute(
+
+            asyncio.create_task(executor_proxy.watch(dispatch_id, node_id, executor))
+
+            output, stdout, stderr, status = await executor._execute(
                 function=assembled_callable,
                 args=inputs["args"],
                 kwargs=inputs["kwargs"],
@@ -296,10 +335,6 @@ async def _run_task(
                 results_dir=results_dir,
                 node_id=node_id,
             )
-            if exception_raised:
-                status = Result.FAILED
-            else:
-                status = Result.COMPLETED
 
             node_result = datasvc.generate_node_result(
                 node_id=node_id,
@@ -450,7 +485,7 @@ async def _postprocess_workflow(result_object: Result) -> Result:
     post_processor = [pp_executor, pp_executor_data]
 
     result_object._status = Result.POSTPROCESSING
-    upsert._lattice_data(result_object)
+    upsert.lattice_data(result_object)
 
     app_log.debug(f"Preparing to post-process workflow {result_object.dispatch_id}")
 
@@ -458,7 +493,7 @@ async def _postprocess_workflow(result_object: Result) -> Result:
         app_log.debug("Workflow to be postprocessed client side")
         result_object._status = Result.PENDING_POSTPROCESSING
         result_object._end_time = datetime.now(timezone.utc)
-        upsert._lattice_data(result_object)
+        upsert.lattice_data(result_object)
         return result_object
 
     post_processing_inputs = {}
@@ -490,7 +525,7 @@ async def _postprocess_workflow(result_object: Result) -> Result:
         result_object._status = Result.POSTPROCESSING_FAILED
         result_object._error = f"Post-processing failed: {error_msg}"
         result_object._end_time = datetime.now(timezone.utc)
-        upsert._lattice_data(result_object)
+        upsert.lattice_data(result_object)
 
         app_log.debug("Returning from _postprocess_workflow")
         return result_object
@@ -507,5 +542,125 @@ async def _postprocess_workflow(result_object: Result) -> Result:
 
 
 async def postprocess_workflow(dispatch_id: str) -> Result:
+    """
+    Public facing `postprocess_workflow`
+
+    Arg(s)
+        dispatch_id: Unique identifier for dispatched workflow
+
+    Return(s)
+        result_object: Result object of the workflow
+    """
     result_object = datasvc.get_result_object(dispatch_id)
     return await _postprocess_workflow(result_object)
+
+
+async def _cancel_task(
+    dispatch_id: str, task_id: int, executor, executor_data: Dict, job_handle: str
+) -> Union[Any, Literal[False]]:
+    """
+    Cancel the task currently being executed by the executor
+
+    Arg(s)
+        dispatch_id: Dispatch ID
+        task_id: Task ID of the electron in transport graph to be cancelled
+        executor: Covalent executor currently being used to execute the task
+        executor_data: Executor configuration arguments
+        job_handle: Unique identifier assigned to the task by the backend running the job
+
+    Return(s)
+        cancel_job_result: Status of the job cancellation action
+    """
+    app_log.debug(f"Cancel task {task_id} using executor {executor}, {executor_data}")
+    app_log.debug(f"job_handle: {job_handle}")
+
+    try:
+        executor = _executor_manager.get_executor(executor)
+        executor.from_dict(executor_data)
+        executor._init_runtime(loop=asyncio.get_running_loop(), cancel_pool=_cancel_threadpool)
+
+        task_metadata = {"dispatch_id": dispatch_id, "node_id": task_id}
+
+        cancel_job_result = await executor._cancel(task_metadata, json.loads(job_handle))
+    except Exception as ex:
+        app_log.debug(f"Execption when cancel task {dispatch_id}:{task_id}: {ex}")
+        cancel_job_result = False
+
+    await set_cancel_result(dispatch_id, task_id, cancel_job_result)
+    return cancel_job_result
+
+
+def to_cancel_kwargs(
+    index: int, node_id: int, node_metadata: List[dict], job_metadata: List[dict]
+) -> dict:
+    """
+    Convert node_metadata for a given node `node_id` into a dictionary
+
+    Arg(s)
+        index: Index into the node_metadata list
+        node_id: Node ID
+        node_metadata: List of node metadata attributes
+        job_metadata: List of metadata for the current job
+
+    Return(s)
+        Node metadata dictionary
+    """
+    return {
+        "task_id": node_id,
+        "executor": node_metadata[index]["executor"],
+        "executor_data": node_metadata[index]["executor_data"],
+        "job_handle": job_metadata[index]["job_handle"],
+    }
+
+
+async def cancel_tasks(dispatch_id: str, task_ids: List[int]) -> None:
+    """
+    Request all tasks with `task_ids` to be cancelled in the workflow identified by `dispatch_id`
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the workflow
+        task_ids: List of task ids to be cancelled
+
+    Return(s)
+        None
+    """
+    job_metadata = await get_jobs_metadata(dispatch_id, task_ids)
+    node_metadata = _get_metadata_for_nodes(dispatch_id, task_ids)
+
+    cancel_task_kwargs = [
+        to_cancel_kwargs(i, x, node_metadata, job_metadata) for i, x in enumerate(task_ids)
+    ]
+
+    for kwargs in cancel_task_kwargs:
+        asyncio.create_task(_cancel_task(dispatch_id, **kwargs))
+
+
+def _get_metadata_for_nodes(dispatch_id: str, node_ids: list) -> List[Any]:
+    """
+    Returns all the metadata associated with the node(s) for the workflow identified by `dispatch_id`
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the workflow
+        node_ids: List of node ids from the workflow to retrive the metadata for
+
+    Return(s)
+        List of node metadata for the given `node_ids`
+    """
+    res = datasvc.get_result_object(dispatch_id)
+    tg = res.lattice.transport_graph
+    return list(map(lambda x: tg.get_node_value(x, "metadata"), node_ids))
+
+
+async def _get_cancel_requested(dispatch_id: str, task_id: int) -> Any:
+    """
+    Query if a specific task has been requested to be cancelled
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the workflow
+        task_id: ID of the node to be cancelled
+
+    Return(s)
+        Whether the task has been requested to be cancelled or not
+    """
+    records = await get_jobs_metadata(dispatch_id, [task_id])
+    return records[0]["cancel_requested"]
diff --git a/covalent_dispatcher/_core/runner_modules/__init__.py b/covalent_dispatcher/_core/runner_modules/__init__.py
new file mode 100644
index 00000000..523f7762
--- /dev/null
+++ b/covalent_dispatcher/_core/runner_modules/__init__.py
@@ -0,0 +1,19 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
diff --git a/covalent_dispatcher/_core/runner_modules/executor_proxy.py b/covalent_dispatcher/_core/runner_modules/executor_proxy.py
new file mode 100644
index 00000000..2cf6e83b
--- /dev/null
+++ b/covalent_dispatcher/_core/runner_modules/executor_proxy.py
@@ -0,0 +1,140 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+""" Monitor executor instances """
+
+
+from typing import Any
+
+from covalent._shared_files import logger
+from covalent.executor.base import _AbstractBaseExecutor as _ABE
+from covalent.executor.utils import Signals
+
+from ..data_modules import job_manager
+
+app_log = logger.app_log
+log_stack_info = logger.log_stack_info
+
+_putters = {}
+_getters = {}
+
+
+async def _get_cancel_requested(dispatch_id: str, task_id: int):
+    """
+    Query the database for the task's cancellation status
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+        task_id: ID of the task within the lattice
+
+    Return(s)
+        Canellation status of the task
+    """
+    # Don't hit the DB for post-processing task
+    if task_id < 0:
+        return False
+
+    app_log.debug(f"Get _handle_requested for executor {dispatch_id}:{task_id}")
+    job_records = await job_manager.get_jobs_metadata(dispatch_id, [task_id])
+    app_log.debug(f"Job record: {job_records[0]}")
+    return job_records[0]["cancel_requested"]
+
+
+async def _put_job_handle(dispatch_id: str, task_id: int, job_handle: str) -> bool:
+    """
+    Store the job handle of the task returned by the backend in the database
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+        task_id: ID of the task within the lattice
+        job_handle: Unique identifier of the task returned by the execution backend
+
+    Return(s)
+        True
+    """
+    # Don't hit the DB for post-processing task
+    if task_id < 0:
+        return False
+    app_log.debug(f"Put job_handle for executor {dispatch_id}:{task_id}")
+    await job_manager.set_job_handle(dispatch_id, task_id, job_handle)
+    return True
+
+
+_putters["job_handle"] = _put_job_handle
+_getters["cancel_requested"] = _get_cancel_requested
+
+
+async def _handle_message(
+    dispatch_id: str, task_id: int, executor: _ABE, action: Signals, body: Any = None
+) -> Any:
+    """
+    Handle the action properly
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+        task_id: ID of the task in the lattice
+        executor: Instance of the abstract base executor
+        action: Action requested to be performed
+        body: Content of the action/request made
+
+    Return(s)
+        Response corresponding to the action requested
+    """
+    if action == Signals.GET:
+        return await _getters[body](dispatch_id, task_id)
+
+    if action == Signals.PUT:
+        key, val = body
+        await _putters[key](dispatch_id, task_id, val)
+        return None
+    else:
+        raise KeyError(f"Unknown action {action}")
+
+
+async def watch(dispatch_id: str, task_id: int, executor: _ABE) -> None:
+    """
+    Watch the send and receive queues of the executor
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+        task_id: ID of the task within the lattice
+        executor: Instance of the abstract base executor
+
+    Return(s)
+        None
+    """
+    send_queue = executor._send_queue
+    recv_queue = executor._recv_queue
+
+    while True:
+        action, body = await send_queue.get()
+        app_log.debug(f"Received message {action} from executor {dispatch_id}:{task_id}")
+
+        if action == Signals.EXIT:
+            app_log.debug(f"Stopping listener for executor {dispatch_id}:{task_id}")
+            break
+
+        try:
+            resp = await _handle_message(dispatch_id, task_id, executor, action, body)
+            recv_queue.put_nowait((True, resp))
+        except Exception as ex:
+            app_log.warning(f"Error handling message {action} from executor: {ex}")
+            recv_queue.put_nowait((False, None))
+            break
diff --git a/covalent_dispatcher/_db/jobdb.py b/covalent_dispatcher/_db/jobdb.py
new file mode 100644
index 00000000..712d5988
--- /dev/null
+++ b/covalent_dispatcher/_db/jobdb.py
@@ -0,0 +1,170 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""This module contains all the functions required interface with the jobs table"""
+
+from typing import Dict, List
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from covalent._shared_files import logger
+
+from .datastore import workflow_db
+from .models import Electron, Job, Lattice
+
+app_log = logger.app_log
+log_stack_info = logger.log_stack_info
+
+
+class MissingJobRecordError(Exception):
+    """
+    Exception to be raised when the job record associated with a particular task is not found in the database
+    """
+
+    def __init__(self, message: str = ""):
+        self.message = message
+        super().__init__(self.message)
+
+
+def transaction_get_job_record(session: Session, job_id: int) -> Dict:
+    """
+    Query the database for the job record associated with the given job id using the passed in session
+    Session object is passed to ensure that for all db related calls the same session is used
+
+    Arg(s)
+        session: SQLalchemy session object
+        job_id: ID of the job to query
+
+    Return(s)
+        Dictionary of the job record from the database
+    """
+    if job_record := session.query(Job).where(Job.id == job_id).first():
+        return {
+            "job_id": job_record.id,
+            "cancel_requested": job_record.cancel_requested,
+            "cancel_successful": job_record.cancel_successful,
+            "job_handle": job_record.job_handle,
+        }
+    else:
+        raise MissingJobRecordError(message=f"Job {job_id} not found")
+
+
+def _update_job_record(
+    session: Session,
+    job_id: int,
+    cancel_requested: bool = None,
+    cancel_successful: bool = None,
+    job_handle: str = None,
+):
+    """
+    Update the job record in the database
+
+    Arg(s)
+        session: SQLalchemy session object
+        job_id: ID of the job to update
+        cancel_requested: Boolean flag indicating whether the job was requested to be cancelled
+        cancel_successful: Boolean indicating whether the job was cancelled successfully
+        job_handle: Unique job handle returned by the execution backend
+
+    Return(s)
+        None
+    """
+    job_record = session.query(Job).where(Job.id == job_id).first()
+    if not job_record:
+        raise MissingJobRecordError(message=f"Job {job_id} not found")
+
+    if cancel_requested is not None:
+        job_record.cancel_requested = cancel_requested
+    if cancel_successful is not None:
+        job_record.cancel_successful = cancel_successful
+    if job_handle is not None:
+        job_record.job_handle = job_handle
+
+
+def get_job_record(job_id: int) -> Dict:
+    """
+    Retrive the job record from database
+
+    Arg(s)
+        job_id: ID of the job to be returned
+
+    Return(s)
+        Job record of the job identified by `job_id`
+    """
+    return get_job_records([job_id])[0]
+
+
+def update_job_records(record_kwargs_list: list):
+    """
+    Update job records in the database
+
+    Arg(s)
+        record_kwargs_list: List of keyword arguments of the fields that need to be updated in the job records
+
+    Return(s)
+        None
+    """
+    with workflow_db.session() as session:
+        for entry in record_kwargs_list:
+            _update_job_record(session, **entry)
+
+
+def get_job_records(job_ids: List[int]) -> List[Dict]:
+    """
+    Retrieve the job records of all jobs with `job_ids`
+
+    Arg(s)
+        job_ids: List of job ids to query the job records of
+
+    Return(s)
+        Job records of all tasks with `job_ids`
+    """
+    with workflow_db.session() as session:
+        records = list(map(lambda x: transaction_get_job_record(session, x), job_ids))
+    return records
+
+
+def to_job_ids(dispatch_id: str, task_ids: List[int]) -> List[int]:
+    """
+    Map all lattice task ids to their corresponding job ids
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+        task_ids: IDs of tasks in the lattice
+
+    Return(s)
+        Corresponding job ids assocated with the provided task ids
+    """
+    with workflow_db.session() as session:
+        stmt = select(Lattice).where(Lattice.dispatch_id == dispatch_id)
+        lattice_rec = session.scalars(stmt).first()
+        if not lattice_rec:
+            raise KeyError(f"Invalid dispatch {dispatch_id}")
+
+        stmt = (
+            select(Electron.job_id)
+            .where(Electron.parent_lattice_id == lattice_rec.id)
+            .where(Electron.transport_graph_node_id.in_(task_ids))
+        )
+
+        records = session.scalars(stmt).all()
+
+        return records
diff --git a/covalent_dispatcher/_db/models.py b/covalent_dispatcher/_db/models.py
index 80613359..1c4007b5 100644
--- a/covalent_dispatcher/_db/models.py
+++ b/covalent_dispatcher/_db/models.py
@@ -188,6 +188,9 @@ class Electron(Base):
     # Name of the column which signifies soft deletion of the electrons corresponding to a lattice
     is_active = Column(Boolean, nullable=False, default=True)
 
+    # Foreign key reference to Jobs table
+    job_id = Column(Integer, ForeignKey("jobs.id", name="job_id_link"), nullable=False)
+
     # Timestamps
     created_at = Column(DateTime, nullable=False, server_default=func.now())
     updated_at = Column(DateTime, nullable=False, onupdate=func.now(), server_default=func.now())
@@ -200,10 +203,12 @@ class ElectronDependency(Base):
     id = Column(Integer, primary_key=True)
 
     # Unique ID of electron
-    electron_id = Column(Integer, nullable=False)
+    electron_id = Column(Integer, ForeignKey("electrons.id", name="electron_link"), nullable=False)
 
     # Unique ID of the electron's parent
-    parent_electron_id = Column(Integer, nullable=False)
+    parent_electron_id = Column(
+        Integer, ForeignKey("electrons.id", name="electron_link"), nullable=False
+    )
 
     edge_name = Column(Text, nullable=False)
 
@@ -218,3 +223,18 @@ class ElectronDependency(Base):
 
     updated_at = Column(DateTime, nullable=True, onupdate=func.now(), server_default=func.now())
     created_at = Column(DateTime, nullable=False, server_default=func.now())
+
+
+class Job(Base):
+    __tablename__ = "jobs"
+    id = Column(Integer, primary_key=True)
+
+    # Indicates whether the job has been requested to be cancelled
+    cancel_requested = Column(Boolean, nullable=False, default=False)
+
+    # Indicates whether the task cancellation succeeded (return value
+    # of Executor.cancel())
+    cancel_successful = Column(Boolean, nullable=False, default=False)
+
+    # JSON-serialized identifier for job
+    job_handle = Column(Text, nullable=False, default="null")
diff --git a/covalent_dispatcher/_db/update.py b/covalent_dispatcher/_db/update.py
index fa1475d5..f8803ada 100644
--- a/covalent_dispatcher/_db/update.py
+++ b/covalent_dispatcher/_db/update.py
@@ -31,7 +31,6 @@ from covalent._workflow.lattice import Lattice
 from covalent._workflow.transport import _TransportGraph
 
 from . import upsert
-from .write_result_to_db import upsert_electron_dependency_data
 
 app_log = logger.app_log
 
@@ -47,11 +46,8 @@ def persist(record: Union[Result, Lattice, _TransportGraph], electron_id: int =
     """
     if isinstance(record, Result):
         _initialize_results_dir(record)
-        app_log.debug("upsert start")
-        upsert._lattice_data(record, electron_id=electron_id)
-        upsert._electron_data(record)
-        app_log.debug("upsert complete")
-        upsert_electron_dependency_data(record.dispatch_id, record.lattice)
+        app_log.debug(f"Persisting {record}")
+        upsert.persist_result(record, electron_id)
         app_log.debug("persist complete")
     if isinstance(record, Lattice):
         persist(record.transport_graph)
@@ -107,7 +103,7 @@ def _node(
         stderr=stderr,
     )
 
-    upsert._electron_data(result)
+    upsert.electron_data(result)
 
 
 def _initialize_results_dir(result):
diff --git a/covalent_dispatcher/_db/upsert.py b/covalent_dispatcher/_db/upsert.py
index 21497208..8f7e99c2 100644
--- a/covalent_dispatcher/_db/upsert.py
+++ b/covalent_dispatcher/_db/upsert.py
@@ -22,20 +22,24 @@ import os
 from datetime import datetime, timezone
 from pathlib import Path
 
+from sqlalchemy.orm import Session
+
 from covalent._results_manager import Result
 from covalent._shared_files import logger
 from covalent._shared_files.config import get_config
 
 from . import models
 from .datastore import workflow_db
+from .jobdb import transaction_get_job_record
 from .write_result_to_db import (
     get_electron_type,
-    insert_electrons_data,
-    insert_lattices_data,
     store_file,
+    transaction_insert_electrons_data,
+    transaction_insert_lattices_data,
+    transaction_update_lattices_data,
+    transaction_upsert_electron_dependency_data,
     update_electrons_data,
     update_lattice_completed_electron_num,
-    update_lattices_data,
 )
 
 app_log = logger.app_log
@@ -71,14 +75,24 @@ LATTICE_LATTICE_IMPORTS_FILENAME = "lattice_imports.pkl"
 LATTICE_STORAGE_TYPE = "local"
 
 
-def _lattice_data(result: Result, electron_id: int = None):
-    with workflow_db.session() as session:
-        lattice_exists = (
-            session.query(models.Lattice)
-            .where(models.Lattice.dispatch_id == result.dispatch_id)
-            .first()
-            is not None
-        )
+def _lattice_data(session: Session, result: Result, electron_id: int = None) -> None:
+    """
+    Private method to update lattice data in database
+
+    Arg(s)
+        session: SQLalchemy session object
+        result: Result object associated with the lattice
+        electron_id: electron id in the lattice
+
+    Return(s)
+        None
+    """
+    lattice_exists = (
+        session.query(models.Lattice)
+        .where(models.Lattice.dispatch_id == result.dispatch_id)
+        .first()
+        is not None
+    )
 
     try:
         workflow_func_string = result.lattice.workflow_function_string
@@ -147,7 +161,7 @@ def _lattice_data(result: Result, electron_id: int = None):
             "started_at": result.start_time,
             "completed_at": result.end_time,
         }
-        insert_lattices_data(**lattice_record_kwarg)
+        transaction_insert_lattices_data(session=session, **lattice_record_kwarg)
 
     else:
         lattice_record_kwarg = {
@@ -158,133 +172,197 @@ def _lattice_data(result: Result, electron_id: int = None):
             "started_at": result.start_time,
             "completed_at": result.end_time,
         }
-        update_lattices_data(**lattice_record_kwarg)
+        transaction_update_lattices_data(session=session, **lattice_record_kwarg)
+
+
+def _electron_data(session: Session, result: Result, cancel_requested: bool = False):
+    """
+    Update electron data in database
 
+    Arg(s)
+        session: SQLalchemy session object
+        result: Result object associated with the lattice
+        cancel_requested: Boolean indicating whether electron was requested to be cancelled
 
-def _electron_data(result: Result):
-    """Update electron data"""
+    Return(s)
+        None
+    """
     tg = result.lattice.transport_graph
     dirty_nodes = set(tg.dirty_nodes)
     tg.dirty_nodes.clear()  # Ensure that dirty nodes list is reset once the data is updated
-    app_log.debug("upsert_electron_data session begin")
-    with workflow_db.session() as session:
-        app_log.debug("upsert_electron_data session success")
-        for node_id in dirty_nodes:
-            results_dir = os.environ.get("COVALENT_DATA_DIR") or get_config(
-                "dispatcher.results_dir"
-            )
-            node_path = Path(os.path.join(results_dir, result.dispatch_id, f"node_{node_id}"))
-
-            if not node_path.exists():
-                node_path.mkdir()
-
-            node_name = tg.get_node_value(node_key=node_id, value_key="name")
-
-            try:
-                function_string = tg.get_node_value(node_id, "function_string")
-            except KeyError:
-                function_string = None
-            try:
-                node_value = tg.get_node_value(node_id, "value")
-            except KeyError:
-                node_value = None
-            try:
-                node_stdout = tg.get_node_value(node_id, "stdout")
-            except KeyError:
-                node_stdout = None
-            try:
-                node_stderr = tg.get_node_value(node_id, "stderr")
-            except KeyError:
-                node_stderr = None
-            try:
-                node_error = tg.get_node_value(node_id, "error")
-            except KeyError:
-                node_error = None
-            try:
-                node_output = tg.get_node_value(node_id, "output")
-            except KeyError:
-                node_output = None
-
-            executor = tg.get_node_value(node_id, "metadata")["executor"]
-            started_at = tg.get_node_value(node_key=node_id, value_key="start_time")
-            completed_at = tg.get_node_value(node_key=node_id, value_key="end_time")
-
-            for filename, data in [
-                (ELECTRON_FUNCTION_FILENAME, tg.get_node_value(node_id, "function")),
-                (ELECTRON_FUNCTION_STRING_FILENAME, function_string),
-                (ELECTRON_VALUE_FILENAME, node_value),
-                (
-                    ELECTRON_EXECUTOR_DATA_FILENAME,
-                    tg.get_node_value(node_id, "metadata")["executor_data"],
-                ),
-                (ELECTRON_DEPS_FILENAME, tg.get_node_value(node_id, "metadata")["deps"]),
-                (
-                    ELECTRON_CALL_BEFORE_FILENAME,
-                    tg.get_node_value(node_id, "metadata")["call_before"],
-                ),
-                (
-                    ELECTRON_CALL_AFTER_FILENAME,
-                    tg.get_node_value(node_id, "metadata")["call_after"],
-                ),
-                (ELECTRON_STDOUT_FILENAME, node_stdout),
-                (ELECTRON_STDERR_FILENAME, node_stderr),
-                (ELECTRON_ERROR_FILENAME, node_error),
-                (ELECTRON_RESULTS_FILENAME, node_output),
-            ]:
-                store_file(node_path, filename, data)
-
-            electron_exists = (
-                session.query(models.Electron, models.Lattice)
-                .where(
-                    models.Electron.parent_lattice_id == models.Lattice.id,
-                    models.Lattice.dispatch_id == result.dispatch_id,
-                    models.Electron.transport_graph_node_id == node_id,
-                )
-                .first()
-                is not None
+    for node_id in dirty_nodes:
+        results_dir = os.environ.get("COVALENT_DATA_DIR") or get_config("dispatcher.results_dir")
+        node_path = Path(os.path.join(results_dir, result.dispatch_id, f"node_{node_id}"))
+
+        if not node_path.exists():
+            node_path.mkdir()
+
+        node_name = tg.get_node_value(node_id, "name")
+
+        try:
+            function_string = tg.get_node_value(node_id, "function_string")
+        except KeyError:
+            function_string = None
+
+        try:
+            node_value = tg.get_node_value(node_id, "value")
+        except KeyError:
+            node_value = None
+
+        try:
+            node_stdout = tg.get_node_value(node_id, "stdout")
+        except KeyError:
+            node_stdout = None
+
+        try:
+            node_stderr = tg.get_node_value(node_id, "stderr")
+        except KeyError:
+            node_stderr = None
+
+        try:
+            node_error = tg.get_node_value(node_id, "error")
+        except KeyError:
+            node_error = None
+
+        try:
+            node_output = tg.get_node_value(node_id, "output")
+        except KeyError:
+            node_output = None
+
+        executor = tg.get_node_value(node_id, "metadata")["executor"]
+        started_at = tg.get_node_value(node_key=node_id, value_key="start_time")
+        completed_at = tg.get_node_value(node_key=node_id, value_key="end_time")
+
+        for filename, data in [
+            (ELECTRON_FUNCTION_FILENAME, tg.get_node_value(node_id, "function")),
+            (ELECTRON_FUNCTION_STRING_FILENAME, function_string),
+            (ELECTRON_VALUE_FILENAME, node_value),
+            (
+                ELECTRON_EXECUTOR_DATA_FILENAME,
+                tg.get_node_value(node_id, "metadata")["executor_data"],
+            ),
+            (ELECTRON_DEPS_FILENAME, tg.get_node_value(node_id, "metadata")["deps"]),
+            (
+                ELECTRON_CALL_BEFORE_FILENAME,
+                tg.get_node_value(node_id, "metadata")["call_before"],
+            ),
+            (
+                ELECTRON_CALL_AFTER_FILENAME,
+                tg.get_node_value(node_id, "metadata")["call_after"],
+            ),
+            (ELECTRON_STDOUT_FILENAME, node_stdout),
+            (ELECTRON_STDERR_FILENAME, node_stderr),
+            (ELECTRON_ERROR_FILENAME, node_error),
+            (ELECTRON_RESULTS_FILENAME, node_output),
+        ]:
+            store_file(node_path, filename, data)
+
+        electron_exists = (
+            session.query(models.Electron, models.Lattice)
+            .where(
+                models.Electron.parent_lattice_id == models.Lattice.id,
+                models.Lattice.dispatch_id == result.dispatch_id,
+                models.Electron.transport_graph_node_id == node_id,
             )
+            .first()
+            is not None
+        )
+
+        status = tg.get_node_value(node_key=node_id, value_key="status")
+        if not electron_exists:
+            electron_record_kwarg = {
+                "parent_dispatch_id": result.dispatch_id,
+                "transport_graph_node_id": node_id,
+                "type": get_electron_type(tg.get_node_value(node_key=node_id, value_key="name")),
+                "name": node_name,
+                "status": str(status),
+                "storage_type": ELECTRON_STORAGE_TYPE,
+                "storage_path": str(node_path),
+                "function_filename": ELECTRON_FUNCTION_FILENAME,
+                "function_string_filename": ELECTRON_FUNCTION_STRING_FILENAME,
+                "executor": executor,
+                "executor_data_filename": ELECTRON_EXECUTOR_DATA_FILENAME,
+                "results_filename": ELECTRON_RESULTS_FILENAME,
+                "value_filename": ELECTRON_VALUE_FILENAME,
+                "stdout_filename": ELECTRON_STDOUT_FILENAME,
+                "stderr_filename": ELECTRON_STDERR_FILENAME,
+                "error_filename": ELECTRON_ERROR_FILENAME,
+                "deps_filename": ELECTRON_DEPS_FILENAME,
+                "call_before_filename": ELECTRON_CALL_BEFORE_FILENAME,
+                "call_after_filename": ELECTRON_CALL_AFTER_FILENAME,
+                "cancel_requested": cancel_requested,
+                "created_at": datetime.now(timezone.utc),
+                "updated_at": datetime.now(timezone.utc),
+                "started_at": started_at,
+                "completed_at": completed_at,
+            }
+            transaction_insert_electrons_data(session=session, **electron_record_kwarg)
+        else:
+            electron_record_kwarg = {
+                "parent_dispatch_id": result.dispatch_id,
+                "transport_graph_node_id": node_id,
+                "name": node_name,
+                "status": str(status),
+                "started_at": started_at,
+                "updated_at": datetime.now(timezone.utc),
+                "completed_at": completed_at,
+            }
+            update_electrons_data(**electron_record_kwarg)
+            if status == Result.COMPLETED:
+                update_lattice_completed_electron_num(result.dispatch_id)
+
+
+def lattice_data(result: Result, electron_id: int = None) -> None:
+    """
+    Upsert the lattice data to database
+
+    Arg(s)
+        result: Result object associated with lattice
+        electron_id: ID of the electron within the lattice
+
+    Return(s)
+        None
+    """
+    with workflow_db.session() as session:
+        _lattice_data(session, result, electron_id)
 
-            status = tg.get_node_value(node_key=node_id, value_key="status")
-            if not electron_exists:
-                electron_record_kwarg = {
-                    "parent_dispatch_id": result.dispatch_id,
-                    "transport_graph_node_id": node_id,
-                    "type": get_electron_type(
-                        tg.get_node_value(node_key=node_id, value_key="name")
-                    ),
-                    "name": node_name,
-                    "status": str(status),
-                    "storage_type": ELECTRON_STORAGE_TYPE,
-                    "storage_path": str(node_path),
-                    "function_filename": ELECTRON_FUNCTION_FILENAME,
-                    "function_string_filename": ELECTRON_FUNCTION_STRING_FILENAME,
-                    "executor": executor,
-                    "executor_data_filename": ELECTRON_EXECUTOR_DATA_FILENAME,
-                    "results_filename": ELECTRON_RESULTS_FILENAME,
-                    "value_filename": ELECTRON_VALUE_FILENAME,
-                    "stdout_filename": ELECTRON_STDOUT_FILENAME,
-                    "stderr_filename": ELECTRON_STDERR_FILENAME,
-                    "error_filename": ELECTRON_ERROR_FILENAME,
-                    "deps_filename": ELECTRON_DEPS_FILENAME,
-                    "call_before_filename": ELECTRON_CALL_BEFORE_FILENAME,
-                    "call_after_filename": ELECTRON_CALL_AFTER_FILENAME,
-                    "created_at": datetime.now(timezone.utc),
-                    "updated_at": datetime.now(timezone.utc),
-                    "started_at": started_at,
-                    "completed_at": completed_at,
-                }
-                insert_electrons_data(**electron_record_kwarg)
-
-            else:
-                electron_record_kwarg = {
-                    "parent_dispatch_id": result.dispatch_id,
-                    "transport_graph_node_id": node_id,
-                    "name": node_name,
-                    "status": str(status),
-                    "started_at": started_at,
-                    "updated_at": datetime.now(timezone.utc),
-                    "completed_at": completed_at,
-                }
-                update_electrons_data(**electron_record_kwarg)
-                if status == Result.COMPLETED:
-                    update_lattice_completed_electron_num(result.dispatch_id)
+
+def electron_data(result: Result, cancel_requested: bool = False) -> None:
+    """
+    Upsert electron data to the database
+
+    Arg(s)
+        result: Result object associated with the lattice
+        cancel_requested: Boolean indicating whether the electron was requested to be cancelled
+
+    Return(s)
+        None
+    """
+    with workflow_db.session() as session:
+        _electron_data(session, result, cancel_requested)
+
+
+def persist_result(result: Result, electron_id: int = None) -> None:
+    """
+    Persist the result object of the lattice recursively into the database
+
+    Arg(s)
+        result: Result object associated with the lattice
+        electron_id: ID of the electron within the lattice
+
+    Return(s)
+        None
+    """
+    with workflow_db.session() as session:
+        _lattice_data(session, result, electron_id)
+        if electron_id:
+            e_record = (
+                session.query(models.Electron).where(models.Electron.id == electron_id).first()
+            )
+            cancel_requested = transaction_get_job_record(session, e_record.job_id)[
+                "cancel_requested"
+            ]
+        else:
+            cancel_requested = False
+        _electron_data(session, result, cancel_requested)
+        transaction_upsert_electron_dependency_data(session, result.dispatch_id, result.lattice)
diff --git a/covalent_dispatcher/_db/write_result_to_db.py b/covalent_dispatcher/_db/write_result_to_db.py
index 22fed783..7beee4e2 100644
--- a/covalent_dispatcher/_db/write_result_to_db.py
+++ b/covalent_dispatcher/_db/write_result_to_db.py
@@ -29,6 +29,7 @@ from typing import Any
 import cloudpickle
 import networkx as nx
 from sqlalchemy import update
+from sqlalchemy.orm import Session
 
 from covalent._shared_files import logger
 from covalent._shared_files.defaults import (
@@ -46,22 +47,32 @@ from covalent._shared_files.exceptions import MissingLatticeRecordError
 from covalent._workflow.lattice import Lattice as LatticeClass
 
 from .datastore import workflow_db
-from .models import Electron, ElectronDependency, Lattice
+from .models import Electron, ElectronDependency, Job, Lattice
 
 app_log = logger.app_log
 log_stack_info = logger.log_stack_info
 
 
 class MissingElectronRecordError(Exception):
+    """
+    Exception to raise when an electron record is missing from the database
+    """
+
     pass
 
 
 class InvalidFileExtension(Exception):
+    """
+    Exception to raise when an invalid file extension is encountered
+    """
+
     pass
 
 
 def update_lattice_completed_electron_num(dispatch_id: str) -> None:
-    """Update the number of completed electrons by one corresponding to a lattice."""
+    """
+    Update the number of completed electrons by one corresponding to a lattice
+    """
 
     with workflow_db.session() as session:
         session.query(Lattice).filter_by(dispatch_id=dispatch_id).update(
@@ -70,10 +81,10 @@ def update_lattice_completed_electron_num(dispatch_id: str) -> None:
                 "updated_at": dt.now(timezone.utc),
             }
         )
-        session.commit()
 
 
-def insert_lattices_data(
+def transaction_insert_lattices_data(
+    session: Session,
     dispatch_id: str,
     electron_id: int,
     name: str,
@@ -107,7 +118,12 @@ def insert_lattices_data(
     started_at: dt,
     completed_at: dt,
 ) -> int:
-    """This function writes the lattice data / metadata to the Lattices table in the DB."""
+    """
+    This function writes the lattice data / metadata to the Lattices table in the DB
+
+    Return(s)
+        ID of the lattice updated
+    """
 
     lattice_row = Lattice(
         dispatch_id=dispatch_id,
@@ -145,16 +161,27 @@ def insert_lattices_data(
         completed_at=completed_at,
     )
 
-    with workflow_db.session() as session:
-        session.add(lattice_row)
-        lattice_id = lattice_row.id
-        session.commit()
+    session.add(lattice_row)
+    lattice_id = lattice_row.id
 
     app_log.debug(f"returning lattice id {lattice_id}")
     return lattice_id
 
 
-def insert_electrons_data(
+def insert_lattices_data(*args, **kwargs):
+    """
+    Execute the transaction to update lattice data in the database
+
+    Return(s)
+        None
+    """
+    with workflow_db.session() as session:
+        transaction_insert_lattices_data(session, *args, **kwargs)
+    app_log.debug(f"Added lattice record {locals()} to DB")
+
+
+def transaction_insert_electrons_data(
+    session: Session,
     parent_dispatch_id: str,
     transport_graph_node_id: int,
     type: str,
@@ -174,138 +201,184 @@ def insert_electrons_data(
     deps_filename: str,
     call_before_filename: str,
     call_after_filename: str,
+    cancel_requested: bool,
     created_at: dt,
     updated_at: dt,
     started_at: dt,
     completed_at: dt,
 ) -> id:
-    """This function writes the transport graph node data to the Electrons table in the DB."""
+    """
+    This function writes the transport graph node data to the Electrons table in the DB
+
+    Return(s)
+        electron id
+    """
 
     # Check that the foreign key corresponding to this table exists
-    app_log.debug("insert_electrons_data")
-    with workflow_db.session() as session:
-        row = session.query(Lattice).where(Lattice.dispatch_id == parent_dispatch_id).all()
-        if len(row) == 0:
-            raise MissingLatticeRecordError
 
-        parent_lattice_id = row[0].id
-        app_log.debug(parent_lattice_id)
-
-        electron_row = Electron(
-            parent_lattice_id=parent_lattice_id,
-            transport_graph_node_id=transport_graph_node_id,
-            type=type,
-            name=name,
-            status=status,
-            storage_type=storage_type,
-            storage_path=storage_path,
-            function_filename=function_filename,
-            function_string_filename=function_string_filename,
-            executor=executor,
-            executor_data_filename=executor_data_filename,
-            results_filename=results_filename,
-            value_filename=value_filename,
-            stdout_filename=stdout_filename,
-            stderr_filename=stderr_filename,
-            error_filename=error_filename,
-            deps_filename=deps_filename,
-            call_before_filename=call_before_filename,
-            call_after_filename=call_after_filename,
-            is_active=True,
-            created_at=created_at,
-            updated_at=updated_at,
-            started_at=started_at,
-            completed_at=completed_at,
-        )
+    row = session.query(Lattice).where(Lattice.dispatch_id == parent_dispatch_id).all()
+    if len(row) == 0:
+        raise MissingLatticeRecordError
 
-    with workflow_db.session() as session:
-        session.add(electron_row)
-        session.flush()
-        electron_id = electron_row.id
-        session.commit()
+    parent_lattice_id = row[0].id
+
+    job_row = Job(cancel_requested=cancel_requested)
+    session.add(job_row)
+    session.flush()
+
+    electron_row = Electron(
+        parent_lattice_id=parent_lattice_id,
+        transport_graph_node_id=transport_graph_node_id,
+        type=type,
+        name=name,
+        status=status,
+        storage_type=storage_type,
+        storage_path=storage_path,
+        function_filename=function_filename,
+        function_string_filename=function_string_filename,
+        executor=executor,
+        executor_data_filename=executor_data_filename,
+        results_filename=results_filename,
+        value_filename=value_filename,
+        stdout_filename=stdout_filename,
+        stderr_filename=stderr_filename,
+        error_filename=error_filename,
+        deps_filename=deps_filename,
+        call_before_filename=call_before_filename,
+        call_after_filename=call_after_filename,
+        is_active=True,
+        job_id=job_row.id,
+        created_at=created_at,
+        updated_at=updated_at,
+        started_at=started_at,
+        completed_at=completed_at,
+    )
+
+    session.add(electron_row)
+    session.flush()
+    electron_id = electron_row.id
 
     return electron_id
 
 
-def insert_electron_dependency_data(dispatch_id: str, lattice: LatticeClass):
-    """Extract electron dependencies from the lattice transport graph and add them to the DB."""
+def insert_electrons_data(*args, **kwargs):
+    """
+    Execute the electron update SQLalchemy transaction
+
+    Return(s)
+        electron_id
+    """
+    with workflow_db.session() as session:
+        app_log.debug(f"Adding electron {locals()} to DB")
+        return transaction_insert_electrons_data(session, *args, **kwargs)
+
+
+def transaction_insert_electron_dependency_data(
+    session: Session, dispatch_id: str, lattice: LatticeClass
+):
+    """
+    Extract electron dependencies from the lattice transport graph and add them to the DB
+
+    Return(s)
+        dependency information of an electron
+    """
 
     # TODO - Update how we access the transport graph edges directly in favor of using some interface provied by the TransportGraph class.
     node_links = nx.readwrite.node_link_data(lattice.transport_graph._graph)["links"]
 
     electron_dependency_ids = []
-    with workflow_db.session() as session:
-        for edge_data in node_links:
-            electron_id = (
-                session.query(Lattice, Electron)
-                .filter(Lattice.id == Electron.parent_lattice_id)
-                .filter(Lattice.dispatch_id == dispatch_id)
-                .filter(Electron.transport_graph_node_id == edge_data["target"])
-                .first()
-                .Electron.id
-            )
-            parent_electron_id = (
-                session.query(Lattice, Electron)
-                .filter(Lattice.id == Electron.parent_lattice_id)
-                .filter(Lattice.dispatch_id == dispatch_id)
-                .filter(Electron.transport_graph_node_id == edge_data["source"])
-                .first()
-                .Electron.id
-            )
 
-            electron_dependency_row = ElectronDependency(
-                electron_id=electron_id,
-                parent_electron_id=parent_electron_id,
-                edge_name=edge_data["edge_name"],
-                parameter_type=edge_data["param_type"] if "param_type" in edge_data else None,
-                arg_index=edge_data["arg_index"] if "arg_index" in edge_data else None,
-                is_active=True,
-                created_at=dt.now(timezone.utc),
-                updated_at=dt.now(timezone.utc),
-            )
+    for edge_data in node_links:
+        electron_id = (
+            session.query(Lattice, Electron)
+            .filter(Lattice.id == Electron.parent_lattice_id)
+            .filter(Lattice.dispatch_id == dispatch_id)
+            .filter(Electron.transport_graph_node_id == edge_data["target"])
+            .first()
+            .Electron.id
+        )
+        parent_electron_id = (
+            session.query(Lattice, Electron)
+            .filter(Lattice.id == Electron.parent_lattice_id)
+            .filter(Lattice.dispatch_id == dispatch_id)
+            .filter(Electron.transport_graph_node_id == edge_data["source"])
+            .first()
+            .Electron.id
+        )
 
-            session.add(electron_dependency_row)
-            electron_dependency_ids.append(electron_dependency_row.id)
-        session.commit()
+        electron_dependency_row = ElectronDependency(
+            electron_id=electron_id,
+            parent_electron_id=parent_electron_id,
+            edge_name=edge_data["edge_name"],
+            parameter_type=edge_data["param_type"] if "param_type" in edge_data else None,
+            arg_index=edge_data["arg_index"] if "arg_index" in edge_data else None,
+            is_active=True,
+            created_at=dt.now(timezone.utc),
+            updated_at=dt.now(timezone.utc),
+        )
+
+        session.add(electron_dependency_row)
+        electron_dependency_ids.append(electron_dependency_row.id)
 
     return electron_dependency_ids
 
 
-def upsert_electron_dependency_data(dispatch_id: str, lattice: LatticeClass):
-    """Update electron dependency data"""
+def insert_electron_dependency_data(*args, **kwargs):
+    """
+    persist the electron dependency information into the database
 
-    # Insert electron dependency records if they don't exist
+    Return(s)
+        None
+    """
     with workflow_db.session() as session:
-        electron_dependencies_exist = (
-            session.query(ElectronDependency, Electron, Lattice)
-            .where(
-                Electron.id == ElectronDependency.electron_id,
-                Electron.parent_lattice_id == Lattice.id,
-                Lattice.dispatch_id == dispatch_id,
-            )
-            .first()
-            is not None
+        app_log.debug(f"Adding electron dependency data {locals()} to DB")
+        transaction_insert_electron_dependency_data(session, *args, **kwargs)
+
+
+def transaction_upsert_electron_dependency_data(
+    session: Session, dispatch_id: str, lattice: LatticeClass
+):
+    """
+    Insert electron dependency records if they don't exist
+    """
+
+    electron_dependencies_exist = (
+        session.query(ElectronDependency, Electron, Lattice)
+        .where(
+            Electron.id == ElectronDependency.electron_id,
+            Electron.parent_lattice_id == Lattice.id,
+            Lattice.dispatch_id == dispatch_id,
         )
-    app_log.debug("electron_dependencies_exist is " + str(electron_dependencies_exist))
+        .first()
+        is not None
+    )
+    app_log.debug(f"electron_dependencies_exist is {electron_dependencies_exist}")
     if not electron_dependencies_exist:
-        insert_electron_dependency_data(dispatch_id=dispatch_id, lattice=lattice)
+        transaction_insert_electron_dependency_data(
+            session=session, dispatch_id=dispatch_id, lattice=lattice
+        )
 
 
-def update_lattices_data(dispatch_id: str, **kwargs) -> None:
+def transaction_update_lattices_data(session: Session, dispatch_id: str, **kwargs) -> None:
     """This function updates the lattices record."""
 
-    with workflow_db.session() as session:
-        valid_update = session.query(Lattice).where(Lattice.dispatch_id == dispatch_id).first()
+    valid_update = session.query(Lattice).where(Lattice.dispatch_id == dispatch_id).first()
 
-        if not valid_update:
-            raise MissingLatticeRecordError
+    if not valid_update:
+        raise MissingLatticeRecordError
+
+    for attr, value in kwargs.items():
+        if value:
+            setattr(valid_update, attr, value)
+
+    session.add(valid_update)
 
-        for attr, value in kwargs.items():
-            if value:
-                setattr(valid_update, attr, value)
 
-        session.add(valid_update)
-        session.commit()
+def update_lattices_data(dispatch_id: str, **kwargs) -> None:
+    """This function updates the lattices record."""
+
+    with workflow_db.session() as session:
+        transaction_update_lattices_data(session, dispatch_id, **kwargs)
 
 
 def update_electrons_data(
@@ -349,7 +422,6 @@ def update_electrons_data(
                 completed_at=completed_at,
             )
         )
-        session.commit()
 
 
 def get_electron_type(node_name: str) -> str:
@@ -384,6 +456,12 @@ def get_electron_type(node_name: str) -> str:
 
 
 def get_sublattice_electron_id(parent_dispatch_id: str, sublattice_node_id: int):
+    """
+    Query the electron ID is a sublattice
+
+    Return(s)
+        sublattice electron id
+    """
     with workflow_db.session() as session:
         sublattice_electron_id = (
             session.query(Lattice, Electron)
@@ -400,8 +478,12 @@ def get_sublattice_electron_id(parent_dispatch_id: str, sublattice_node_id: int)
 
 
 def resolve_electron_id(eid: int):
-    """Given an electron's unique id, return the corresponding
-    dispatch_id and node_id
+    """
+    Given an electron's unique id, return the corresponding dispatch_id and node_id
+
+    Return(s)
+        dispatch_id of the lattice
+        ID of the node within the lattice
     """
     with workflow_db.session() as session:
         row = (
@@ -415,7 +497,17 @@ def resolve_electron_id(eid: int):
     return dispatch_id, node_id
 
 
-def write_lattice_error(dispatch_id: str, error: str):
+def write_lattice_error(dispatch_id: str, error: str) -> None:
+    """
+    Persist the lattice error into the database
+
+    Arg(s)
+        dispatch_id: Dispatch ID of the lattice
+        error: Lattice error to be persisted
+
+    Return(s)
+        None
+    """
     with workflow_db.session() as session:
         valid_update = session.query(Lattice).where(Lattice.dispatch_id == dispatch_id).first()
 
diff --git a/covalent_dispatcher/_service/app.py b/covalent_dispatcher/_service/app.py
index 983ca7a6..823f55c3 100644
--- a/covalent_dispatcher/_service/app.py
+++ b/covalent_dispatcher/_service/app.py
@@ -58,7 +58,6 @@ async def submit(request: Request) -> UUID:
     try:
         data = await request.json()
         data = json.dumps(data).encode("utf-8")
-
         return await dispatcher.run_dispatcher(data)
     except Exception as e:
         raise HTTPException(
@@ -103,17 +102,17 @@ async def cancel(request: Request) -> str:
         Fast API Response object confirming that the dispatch
         has been cancelled.
     """
-    try:
-        data = await request.body()
-        dispatch_id = data.decode("utf-8")
 
-        dispatcher.cancel_running_dispatch(dispatch_id)
+    data = await request.json()
+
+    dispatch_id = data["dispatch_id"]
+    task_ids = data["task_ids"]
+
+    await dispatcher.cancel_running_dispatch(dispatch_id, task_ids)
+    if task_ids:
+        return f"Cancelled tasks {task_ids} in dispatch {dispatch_id}."
+    else:
         return f"Dispatch {dispatch_id} cancelled."
-    except Exception as e:
-        raise HTTPException(
-            status_code=400,
-            detail=f"Failed to cancel workflow: {e}",
-        ) from e
 
 
 @router.get("/result/{dispatch_id}")
diff --git a/covalent_dispatcher/entry_point.py b/covalent_dispatcher/entry_point.py
index e4813b12..5e27cc90 100644
--- a/covalent_dispatcher/entry_point.py
+++ b/covalent_dispatcher/entry_point.py
@@ -22,9 +22,12 @@
 Self-contained entry point for the dispatcher
 """
 
+from typing import List
 
 from covalent._shared_files import logger
 
+from ._core import cancel_dispatch
+
 app_log = logger.app_log
 log_stack_info = logger.log_stack_info
 
@@ -67,7 +70,7 @@ async def run_redispatch(
     return redispatch_id
 
 
-def cancel_running_dispatch(dispatch_id: str) -> None:
+async def cancel_running_dispatch(dispatch_id: str, task_ids: List[int] = None) -> None:
     """
     Cancels a running dispatch job.
 
@@ -78,6 +81,7 @@ def cancel_running_dispatch(dispatch_id: str) -> None:
         None
     """
 
-    from ._core import cancel_workflow
+    if task_ids is None:
+        task_ids = []
 
-    cancel_workflow(dispatch_id)
+    await cancel_dispatch(dispatch_id, task_ids)
diff --git a/covalent_migrations/versions/f64ecaa040d5_add_jobs_db_table.py b/covalent_migrations/versions/f64ecaa040d5_add_jobs_db_table.py
new file mode 100644
index 00000000..de23194f
--- /dev/null
+++ b/covalent_migrations/versions/f64ecaa040d5_add_jobs_db_table.py
@@ -0,0 +1,72 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""Add jobs db table
+
+Revision ID: f64ecaa040d5
+Revises: 97202f5f47cb
+Create Date: 2023-02-07 14:52:43.475708
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+# pragma: allowlist nextline secret
+revision = "f64ecaa040d5"
+# pragma: allowlist nextline secret
+down_revision = "97202f5f47cb"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "jobs",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("cancel_requested", sa.Boolean(), nullable=False),
+        sa.Column("cancel_successful", sa.Boolean(), nullable=False),
+        sa.Column("job_handle", sa.Text(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    with op.batch_alter_table("electron_dependency", schema=None) as batch_op:
+        batch_op.create_foreign_key("electron_link", "electrons", ["parent_electron_id"], ["id"])
+        batch_op.create_foreign_key("electron_link", "electrons", ["electron_id"], ["id"])
+
+    with op.batch_alter_table("electrons", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("job_id", sa.Integer(), nullable=False))
+        batch_op.create_foreign_key("job_id_link", "jobs", ["job_id"], ["id"])
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("electrons", schema=None) as batch_op:
+        batch_op.drop_constraint("job_id_link", type_="foreignkey")
+        batch_op.drop_column("job_id")
+
+    with op.batch_alter_table("electron_dependency", schema=None) as batch_op:
+        batch_op.drop_constraint("electron_link", type_="foreignkey")
+        batch_op.drop_constraint("electron_link", type_="foreignkey")
+
+    op.drop_table("jobs")
+    # ### end Alembic commands ###


Test patch:
diff --git a/tests/covalent_dispatcher_tests/_core/data_manager_test.py b/tests/covalent_dispatcher_tests/_core/data_manager_test.py
index 1499b741..41b43d50 100644
--- a/tests/covalent_dispatcher_tests/_core/data_manager_test.py
+++ b/tests/covalent_dispatcher_tests/_core/data_manager_test.py
@@ -298,6 +298,9 @@ def test_make_derived_dispatch_from_old_result(mocker, reuse):
 
 
 def test_get_result_object(mocker):
+    """
+    Test get result object
+    """
     result_object = get_mock_result()
     dispatch_id = result_object.dispatch_id
     _registered_dispatches[dispatch_id] = result_object
@@ -306,6 +309,9 @@ def test_get_result_object(mocker):
 
 
 def test_register_result_object(mocker):
+    """
+    Test registering a result object
+    """
     result_object = get_mock_result()
     dispatch_id = result_object.dispatch_id
     _register_result_object(result_object)
@@ -314,6 +320,9 @@ def test_register_result_object(mocker):
 
 
 def test_unregister_result_object(mocker):
+    """
+    Test unregistering a result object from lattice
+    """
     result_object = get_mock_result()
     dispatch_id = result_object.dispatch_id
     _registered_dispatches[dispatch_id] = result_object
@@ -322,6 +331,9 @@ def test_unregister_result_object(mocker):
 
 
 def test_get_status_queue():
+    """
+    Test querying the dispatch status from the queue
+    """
     import asyncio
 
     dispatch_id = "dispatch"
@@ -332,6 +344,9 @@ def test_get_status_queue():
 
 @pytest.mark.asyncio
 async def test_persist_result(mocker):
+    """
+    Test persisting the result object
+    """
     result_object = get_mock_result()
 
     mock_get_result = mocker.patch(
@@ -353,6 +368,9 @@ async def test_persist_result(mocker):
 )
 @pytest.mark.asyncio
 async def test_update_parent_electron(mocker, sub_status, mapped_status):
+    """
+    Test updating parent electron data
+    """
     parent_result_obj = get_mock_result()
     sub_result_obj = get_mock_result()
     eid = 5
@@ -370,13 +388,13 @@ async def test_update_parent_electron(mocker, sub_status, mapped_status):
         "error": sub_result_obj._error,
     }
 
-    mock_gen_node_result = mocker.patch(
+    mocker.patch(
         "covalent_dispatcher._core.data_manager.generate_node_result",
         return_value=mock_node_result,
     )
 
     mock_update_node = mocker.patch("covalent_dispatcher._core.data_manager.update_node_result")
-    mock_resolve_eid = mocker.patch(
+    mocker.patch(
         "covalent_dispatcher._core.data_manager.resolve_electron_id",
         return_value=(parent_dispatch_id, parent_node_id),
     )
@@ -391,10 +409,13 @@ async def test_update_parent_electron(mocker, sub_status, mapped_status):
 
 
 def test_upsert_lattice_data(mocker):
+    """
+    Test updating lattice data in database
+    """
     result_object = get_mock_result()
-    mock_get_result = mocker.patch(
+    mocker.patch(
         "covalent_dispatcher._core.data_manager.get_result_object", return_value=result_object
     )
-    mock_upsert_lattice = mocker.patch("covalent_dispatcher._db.upsert._lattice_data")
+    mock_upsert_lattice = mocker.patch("covalent_dispatcher._db.upsert.lattice_data")
     upsert_lattice_data(result_object.dispatch_id)
     mock_upsert_lattice.assert_called_with(result_object)
diff --git a/tests/covalent_dispatcher_tests/_core/data_modules/job_manager_test.py b/tests/covalent_dispatcher_tests/_core/data_modules/job_manager_test.py
new file mode 100644
index 00000000..e363dc02
--- /dev/null
+++ b/tests/covalent_dispatcher_tests/_core/data_modules/job_manager_test.py
@@ -0,0 +1,119 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+from functools import partial
+
+import pytest
+
+from covalent_dispatcher._core.data_modules.job_manager import (
+    get_jobs_metadata,
+    set_cancel_requested,
+    set_cancel_result,
+    set_job_handle,
+)
+
+
+def to_job_ids(dispatch_id, task_ids, task_job_map):
+    return list(map(lambda x: task_job_map[x], task_ids))
+
+
+@pytest.mark.asyncio
+async def test_get_jobs_metadata(mocker):
+    """
+    Test retrieving jobs metadata
+    """
+    task_ids = [0, 1]
+    task_job_map = {0: 1, 1: 2}
+    mock_to_job_ids = partial(to_job_ids, task_job_map=task_job_map)
+    job_ids = mock_to_job_ids("dispatch", task_ids)
+    mocker.patch("covalent_dispatcher._core.data_modules.job_manager.to_job_ids", mock_to_job_ids)
+    mock_get = mocker.patch("covalent_dispatcher._core.data_modules.job_manager.get_job_records")
+
+    await get_jobs_metadata("dispatch", task_ids)
+
+    mock_get.assert_called_with(job_ids)
+
+
+@pytest.mark.asyncio
+async def test_set_cancel_requested_private(mocker):
+    """
+    Test the private get cancel requested module method
+    """
+    mock_to_job_ids = mocker.patch(
+        "covalent_dispatcher._core.data_modules.job_manager.to_job_ids", return_value=[0, 1]
+    )
+
+    mock_set_cancel_request_private = mocker.patch(
+        "covalent_dispatcher._core.data_modules.job_manager._set_cancel_requested"
+    )
+
+    await set_cancel_requested("dispatch", [0, 1])
+
+    mock_to_job_ids.assert_called_once_with("dispatch", [0, 1])
+    mock_set_cancel_request_private.assert_called_once_with([0, 1])
+
+
+@pytest.mark.asyncio
+async def test_set_cancel_requested(mocker):
+    """
+    Test set cancel requested
+    """
+    mocker.patch(
+        "covalent_dispatcher._core.data_modules.job_manager.to_job_ids", return_value=[0, 1]
+    )
+    mock_update = mocker.patch(
+        "covalent_dispatcher._core.data_modules.job_manager.update_job_records"
+    )
+
+    await set_cancel_requested("dispatch", [0, 1])
+    expected_args = [{"job_id": job_id, "cancel_requested": True} for job_id in [0, 1]]
+    mock_update.assert_called_with(expected_args)
+
+
+@pytest.mark.asyncio
+async def test_set_job_handle(mocker):
+    """
+    Test set job handle method
+    """
+    task_job_map = {0: 1, 1: 2}
+    mock_to_job_ids = partial(to_job_ids, task_job_map=task_job_map)
+    mocker.patch("covalent_dispatcher._core.data_modules.job_manager.to_job_ids", mock_to_job_ids)
+
+    mock_update = mocker.patch(
+        "covalent_dispatcher._core.data_modules.job_manager.update_job_records"
+    )
+
+    await set_job_handle(dispatch_id="dispatch", task_id=0, job_handle="12356")
+    mock_update.assert_called_with([{"job_id": 1, "job_handle": "12356"}])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("cancel_requested", [True, False])
+async def test_set_cancel_result(cancel_requested, mocker):
+    """
+    Test requesting a task to be cancelled
+    """
+    mock_to_job_ids = partial(to_job_ids, task_job_map={0: 1, 1: 2})
+    mocker.patch("covalent_dispatcher._core.data_modules.job_manager.to_job_ids", mock_to_job_ids)
+    mock_update = mocker.patch(
+        "covalent_dispatcher._core.data_modules.job_manager.update_job_records"
+    )
+    await set_cancel_result("dispatch", 0, cancel_status=cancel_requested)
+    mock_update.assert_called_with([{"job_id": 1, "cancel_successful": cancel_requested}])
diff --git a/tests/covalent_dispatcher_tests/_core/dispatcher_test.py b/tests/covalent_dispatcher_tests/_core/dispatcher_test.py
index 965d8258..fae534d4 100644
--- a/tests/covalent_dispatcher_tests/_core/dispatcher_test.py
+++ b/tests/covalent_dispatcher_tests/_core/dispatcher_test.py
@@ -24,6 +24,7 @@ Tests for the core functionality of the dispatcher.
 
 
 from typing import Dict, List
+from unittest.mock import AsyncMock, call
 
 import cloudpickle as pickle
 import pytest
@@ -39,7 +40,7 @@ from covalent_dispatcher._core.dispatcher import (
     _handle_failed_node,
     _plan_workflow,
     _run_planned_workflow,
-    cancel_workflow,
+    cancel_dispatch,
     run_dispatch,
     run_workflow,
 )
@@ -270,6 +271,9 @@ async def test_get_initial_tasks_and_deps(mocker):
 
 @pytest.mark.asyncio
 async def test_run_dispatch(mocker):
+    """
+    Test running a mock dispatch
+    """
     res = get_mock_result()
     mocker.patch(
         "covalent_dispatcher._core.dispatcher.datasvc.get_result_object", return_value=res
@@ -281,6 +285,9 @@ async def test_run_dispatch(mocker):
 
 @pytest.mark.asyncio
 async def test_run_workflow_normal(mocker):
+    """
+    Test a normal workflow execution
+    """
     import asyncio
 
     result_object = get_mock_result()
@@ -304,6 +311,9 @@ async def test_run_workflow_normal(mocker):
 
 @pytest.mark.asyncio
 async def test_run_completed_workflow(mocker):
+    """
+    Test run completed workflow
+    """
     import asyncio
 
     result_object = get_mock_result()
@@ -316,10 +326,10 @@ async def test_run_completed_workflow(mocker):
         "covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch"
     )
     mock_plan = mocker.patch("covalent_dispatcher._core.dispatcher._plan_workflow")
-    mock_run_planned_workflow = mocker.patch(
+    mocker.patch(
         "covalent_dispatcher._core.dispatcher._run_planned_workflow", return_value=result_object
     )
-    mock_persist = mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.persist_result")
+    mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.persist_result")
 
     await run_workflow(result_object)
 
@@ -330,6 +340,9 @@ async def test_run_completed_workflow(mocker):
 
 @pytest.mark.asyncio
 async def test_run_workflow_exception(mocker):
+    """
+    Test any exception raised when running workflow
+    """
     import asyncio
 
     result_object = get_mock_result()
@@ -358,13 +371,14 @@ async def test_run_workflow_exception(mocker):
 
 @pytest.mark.asyncio
 async def test_run_planned_workflow_cancelled_update(mocker):
+    """
+    Test run planned workflow with cancelled update
+    """
     import asyncio
 
     result_object = get_mock_result()
 
-    mock_upsert_lattice = mocker.patch(
-        "covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data"
-    )
+    mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data")
     tasks_left = 1
     initial_nodes = [0]
     pending_deps = {0: 0}
@@ -391,13 +405,14 @@ async def test_run_planned_workflow_cancelled_update(mocker):
 
 @pytest.mark.asyncio
 async def test_run_planned_workflow_failed_update(mocker):
+    """
+    Test run planned workflow with mocking a failed job update
+    """
     import asyncio
 
     result_object = get_mock_result()
 
-    mock_upsert_lattice = mocker.patch(
-        "covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data"
-    )
+    mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data")
     tasks_left = 1
     initial_nodes = [0]
     pending_deps = {0: 0}
@@ -422,5 +437,80 @@ async def test_run_planned_workflow_failed_update(mocker):
     mock_handle_failed.assert_awaited_with(result_object, 0)
 
 
-def test_cancelled_workflow():
-    cancel_workflow("asdf")
+@pytest.mark.asyncio
+async def test_cancel_dispatch(mocker):
+    """Test cancelling a dispatch, including sub-lattices"""
+    res = get_mock_result()
+    sub_res = get_mock_result()
+
+    sub_dispatch_id = "sub_pipeline_workflow"
+    sub_res._dispatch_id = sub_dispatch_id
+
+    def mock_get_result_object(dispatch_id):
+        objs = {res._dispatch_id: res, sub_res._dispatch_id: sub_res}
+        return objs[dispatch_id]
+
+    mock_data_cancel = mocker.patch("covalent_dispatcher._core.dispatcher.set_cancel_requested")
+
+    mock_runner = mocker.patch("covalent_dispatcher._core.dispatcher.runner")
+    mock_runner.cancel_tasks = AsyncMock()
+
+    mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.get_result_object", mock_get_result_object
+    )
+
+    res._initialize_nodes()
+    sub_res._initialize_nodes()
+
+    tg = res.lattice.transport_graph
+    tg.set_node_value(2, "sub_dispatch_id", sub_dispatch_id)
+    sub_tg = sub_res.lattice.transport_graph
+
+    await cancel_dispatch("pipeline_workflow")
+
+    task_ids = list(tg._graph.nodes)
+    sub_task_ids = list(sub_tg._graph.nodes)
+
+    calls = [call("pipeline_workflow", task_ids), call(sub_dispatch_id, sub_task_ids)]
+    mock_data_cancel.assert_has_awaits(calls)
+    mock_runner.cancel_tasks.assert_has_awaits(calls)
+
+
+@pytest.mark.asyncio
+async def test_cancel_dispatch_with_task_ids(mocker):
+    """Test cancelling a dispatch, including sub-lattices and with task ids"""
+    res = get_mock_result()
+    sub_res = get_mock_result()
+
+    sub_dispatch_id = "sub_pipeline_workflow"
+    sub_res._dispatch_id = sub_dispatch_id
+
+    def mock_get_result_object(dispatch_id):
+        objs = {res._dispatch_id: res, sub_res._dispatch_id: sub_res}
+        return objs[dispatch_id]
+
+    mock_data_cancel = mocker.patch("covalent_dispatcher._core.dispatcher.set_cancel_requested")
+
+    mock_runner = mocker.patch("covalent_dispatcher._core.dispatcher.runner")
+    mock_runner.cancel_tasks = AsyncMock()
+
+    mocker.patch(
+        "covalent_dispatcher._core.dispatcher.datasvc.get_result_object", mock_get_result_object
+    )
+    mock_app_log = mocker.patch("covalent_dispatcher._core.dispatcher.app_log.debug")
+
+    res._initialize_nodes()
+    sub_res._initialize_nodes()
+
+    tg = res.lattice.transport_graph
+    tg.set_node_value(2, "sub_dispatch_id", sub_dispatch_id)
+    sub_tg = sub_res.lattice.transport_graph
+    task_ids = list(tg._graph.nodes)
+    sub_task_ids = list(sub_tg._graph.nodes)
+
+    await cancel_dispatch("pipeline_workflow", task_ids)
+
+    calls = [call("pipeline_workflow", task_ids), call(sub_dispatch_id, sub_task_ids)]
+    mock_data_cancel.assert_has_awaits(calls)
+    mock_runner.cancel_tasks.assert_has_awaits(calls)
+    assert mock_app_log.call_count == 2
diff --git a/tests/covalent_dispatcher_tests/_core/execution_test.py b/tests/covalent_dispatcher_tests/_core/execution_test.py
index 2f67ef86..a2da9db8 100644
--- a/tests/covalent_dispatcher_tests/_core/execution_test.py
+++ b/tests/covalent_dispatcher_tests/_core/execution_test.py
@@ -366,12 +366,16 @@ async def test_run_workflow_with_client_side_postprocess(test_db, mocker):
     mocker.patch(
         "covalent_dispatcher._core.data_manager.get_status_queue", return_value=status_queue
     )
+    mocker.patch("covalent_dispatcher._core.runner._gather_deps")
+    mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.upsert_lattice_data")
+    mock_run_abstract_task = mocker.patch("covalent_dispatcher._core.runner._run_abstract_task")
 
     update.persist(result_object)
 
     result_object = await run_workflow(result_object)
     mock_unregister.assert_called_with(result_object.dispatch_id)
     assert result_object.status == Result.PENDING_POSTPROCESSING
+    mock_run_abstract_task.assert_awaited_once()
 
 
 @pytest.mark.asyncio
@@ -391,6 +395,7 @@ async def test_run_workflow_with_failed_postprocess(test_db, mocker):
     mocker.patch(
         "covalent_dispatcher._core.runner.datasvc.get_result_object", return_value=result_object
     )
+    mocker.patch("covalent_dispatcher._core.runner._run_abstract_task")
 
     update.persist(result_object)
 
@@ -398,6 +403,7 @@ async def test_run_workflow_with_failed_postprocess(test_db, mocker):
     mocker.patch(
         "covalent_dispatcher._core.data_manager.get_status_queue", return_value=status_queue
     )
+    mock_run_abstract_task = mocker.patch("covalent_dispatcher._core.runner._run_abstract_task")
 
     def failing_workflow(x):
         assert False
@@ -415,10 +421,14 @@ async def test_run_workflow_with_failed_postprocess(test_db, mocker):
     mock_unregister.assert_called_with(result_object.dispatch_id)
 
     assert result_object.status == Result.POSTPROCESSING_FAILED
+    assert mock_run_abstract_task.call_count == 2
 
 
 @pytest.mark.asyncio
 async def test_run_task_sublattice_handling(test_db, mocker):
+    """
+    Test exception handling with running a sublattice
+    """
     result_object = get_mock_result()
     sub_result_object = get_mock_result()
     sub_result_object._dispatch_id = "sublattice_workflow"
diff --git a/tests/covalent_dispatcher_tests/_core/runner_test.py b/tests/covalent_dispatcher_tests/_core/runner_test.py
index 1a253e8f..231f6f3b 100644
--- a/tests/covalent_dispatcher_tests/_core/runner_test.py
+++ b/tests/covalent_dispatcher_tests/_core/runner_test.py
@@ -23,6 +23,7 @@ Tests for the core functionality of the runner.
 """
 
 
+import json
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -33,13 +34,18 @@ from covalent._workflow.lattice import Lattice
 from covalent_dispatcher._core.data_manager import get_result_object
 from covalent_dispatcher._core.runner import (
     _build_sublattice_graph,
+    _cancel_task,
     _dispatch_sublattice,
     _gather_deps,
+    _get_cancel_requested,
+    _get_metadata_for_nodes,
     _post_process,
     _postprocess_workflow,
     _run_abstract_task,
     _run_task,
+    cancel_tasks,
 )
+from covalent_dispatcher._db import update
 from covalent_dispatcher._db.datastore import DataStore
 
 TEST_RESULTS_DIR = "/tmp/results"
@@ -75,6 +81,7 @@ def get_mock_result() -> Result:
     pipeline.build_graph(x="absolute")
     received_workflow = Lattice.deserialize_from_json(pipeline.serialize_to_json())
     result_object = Result(received_workflow, "pipeline_workflow")
+    result_object._initialize_nodes()
 
     return result_object
 
@@ -135,6 +142,46 @@ async def test_run_abstract_task_exception_handling(mocker):
     assert node_result["status"] == Result.FAILED
 
 
+@pytest.mark.asyncio
+async def test_run_abstract_task_get_cancel_requested(mocker):
+    """Test that get_cancel_requested is properly handled"""
+    mock_result = MagicMock()
+
+    result_object = get_mock_result()
+    inputs = {"args": [], "kwargs": {}}
+    mock_app_log = mocker.patch("covalent_dispatcher._core.runner.app_log.debug")
+    mock_get_result = mocker.patch(
+        "covalent_dispatcher._core.runner.datasvc.get_result_object", return_value=result_object
+    )
+    mock_get_task_input_values = mocker.patch(
+        "covalent_dispatcher._core.runner._get_task_input_values",
+        side_effect=RuntimeError(),
+    )
+    mock_get_cancel_requested = mocker.patch(
+        "covalent_dispatcher._core.runner._get_cancel_requested",
+        return_value=AsyncMock(return_value=True),
+    )
+    mock_generate_node_result = mocker.patch(
+        "covalent_dispatcher._core.runner.datasvc.generate_node_result",
+        return_value=mock_result,
+    )
+
+    node_result = await _run_abstract_task(
+        dispatch_id=result_object.dispatch_id,
+        node_id=0,
+        node_name="test_node",
+        abstract_inputs=inputs,
+        selected_executor=["local", {}],
+        workflow_executor=["local", {}],
+    )
+
+    mock_get_result.assert_called_with(result_object.dispatch_id)
+    mock_get_cancel_requested.assert_awaited_once_with(result_object.dispatch_id, 0)
+    mock_generate_node_result.assert_called()
+    mock_app_log.assert_called_with(f"Don't run cancelled task {result_object.dispatch_id}:0")
+    assert node_result == mock_result
+
+
 @pytest.mark.asyncio
 async def test_run_task_executor_exception_handling(mocker):
     """Test that exceptions from initializing executors are caught"""
@@ -274,6 +321,10 @@ async def test_postprocess_workflow(mocker):
 
 
 def test_build_sublattice_graph():
+    """
+    Test building a sublattice graph
+    """
+
     @ct.electron
     def task(x):
         return x
@@ -305,6 +356,10 @@ def test_build_sublattice_graph():
 
 @pytest.mark.asyncio
 async def test_dispatch_sublattice(test_db, mocker):
+    """
+    Test dispatching a sublattice and assert any exceptions raised
+    """
+
     @ct.electron(executor="local")
     def task(x):
         return x
@@ -316,8 +371,9 @@ async def test_dispatch_sublattice(test_db, mocker):
     mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
     mocker.patch("covalent_dispatcher._db.upsert.workflow_db", test_db)
     mocker.patch("covalent_dispatcher._core.dispatcher.datasvc.finalize_dispatch")
-
+    mocker.patch("covalent_dispatcher._db.jobdb.workflow_db", test_db)
     result_object = get_mock_result()
+    update.persist(result_object)
 
     serialized_callable = ct.TransportableObject(sub_workflow)
     inputs = {"args": [ct.TransportableObject(2)], "kwargs": {}}
@@ -377,3 +433,123 @@ async def test_dispatch_sublattice(test_db, mocker):
             serialized_callable=serialized_callable,
             workflow_executor=["fake_executor", {}],
         )
+
+
+@pytest.mark.asyncio
+async def test__cancel_task(mocker):
+    """
+    Test module private _cancel_task method
+    """
+    mock_executor = AsyncMock()
+    mock_executor.from_dict = MagicMock()
+    mock_executor._init_runtime = MagicMock()
+    mock_executor._cancel = AsyncMock()
+
+    mock_app_log = mocker.patch("covalent_dispatcher._core.runner.app_log.debug")
+    mock_executor_manager = mocker.patch(
+        "covalent_dispatcher._core.runner._executor_manager.get_executor",
+        return_value=mock_executor,
+    )
+    mock_set_cancel_result = mocker.patch("covalent_dispatcher._core.runner.set_cancel_result")
+
+    dispatch_id = "abcd"
+    task_id = 0
+    executor = "mock_executor"
+    executor_data = {}
+    job_handle = "42"
+
+    task_metadata = {"dispatch_id": dispatch_id, "node_id": task_id}
+
+    await _cancel_task(dispatch_id, task_id, executor, executor_data, job_handle)
+
+    assert mock_app_log.call_count == 2
+    mock_executor_manager.assert_called_with(executor)
+    mock_executor.from_dict.assert_called_with(executor_data)
+    mock_executor._init_runtime.assert_called()
+    mock_executor._cancel.assert_called_with(task_metadata, json.loads(job_handle))
+    mock_set_cancel_result.assert_called()
+
+
+@pytest.mark.asyncio
+async def test__cancel_task_exception(mocker):
+    """
+    Test exception raised in module private _cancel task exception
+    """
+    mock_executor = AsyncMock()
+    mock_executor.from_dict = MagicMock()
+    mock_executor._init_runtime = MagicMock()
+    mock_executor._cancel = AsyncMock(side_effect=Exception("cancel"))
+
+    mock_app_log = mocker.patch("covalent_dispatcher._core.runner.app_log.debug")
+    mock_executor_manager = mocker.patch(
+        "covalent_dispatcher._core.runner._executor_manager.get_executor",
+        return_value=mock_executor,
+    )
+    mock_set_cancel_result = mocker.patch("covalent_dispatcher._core.runner.set_cancel_result")
+
+    dispatch_id = "abcd"
+    task_id = 0
+    executor = "mock_executor"
+    executor_data = {}
+    job_handle = "42"
+
+    task_metadata = {"dispatch_id": dispatch_id, "node_id": task_id}
+
+    cancel_result = await _cancel_task(dispatch_id, task_id, executor, executor_data, job_handle)
+    assert mock_app_log.call_count == 3
+    mock_executor_manager.assert_called_with(executor)
+    mock_executor.from_dict.assert_called_with(executor_data)
+    mock_executor._init_runtime.assert_called()
+    mock_executor._cancel.assert_called_with(task_metadata, json.loads(job_handle))
+    assert cancel_result is False
+
+
+@pytest.mark.asyncio
+async def test_cancel_tasks(mocker):
+    """
+    Test cancelling multiple tasks
+    """
+    mock_get_jobs_metadata = mocker.patch(
+        "covalent_dispatcher._core.runner.get_jobs_metadata", return_value=AsyncMock()
+    )
+    mock_get_metadata_for_nodes = mocker.patch(
+        "covalent_dispatcher._core.runner._get_metadata_for_nodes", return_value=MagicMock()
+    )
+
+    dispatch_id = "abcd"
+    task_ids = [0, 1]
+
+    await cancel_tasks(dispatch_id, task_ids)
+
+    mock_get_jobs_metadata.assert_awaited_with(dispatch_id, task_ids)
+    mock_get_metadata_for_nodes.assert_called_with(dispatch_id, task_ids)
+
+
+def test__get_metadata_for_nodes(mocker):
+    """
+    Test module private method for getting nodes metadata
+    """
+    dispatch_id = "abcd"
+    node_ids = [0, 1]
+
+    mock_get_result_object = mocker.patch(
+        "covalent_dispatcher._core.runner.datasvc.get_result_object", return_value=MagicMock()
+    )
+    _get_metadata_for_nodes(dispatch_id, node_ids)
+    mock_get_result_object.assert_called_with(dispatch_id)
+
+
+@pytest.mark.asyncio
+async def test__get_cancel_requested(mocker):
+    """
+    Test module private method for querying if a task was requested to be cancelled
+    """
+    dispatch_id = "abcd"
+    task_id = 0
+    mock_get_jobs_metadata = mocker.patch(
+        "covalent_dispatcher._core.runner.get_jobs_metadata", return_value=AsyncMock()
+    )
+
+    await _get_cancel_requested(dispatch_id, task_id)
+
+    mock_get_jobs_metadata.assert_awaited_with(dispatch_id, [task_id])
diff --git a/tests/covalent_dispatcher_tests/_db/jobdb_test.py b/tests/covalent_dispatcher_tests/_db/jobdb_test.py
new file mode 100644
index 00000000..bae861cc
--- /dev/null
+++ b/tests/covalent_dispatcher_tests/_db/jobdb_test.py
@@ -0,0 +1,136 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+from datetime import datetime as dt
+from datetime import timezone
+
+import pytest
+
+from covalent_dispatcher._db.datastore import DataStore
+from covalent_dispatcher._db.jobdb import (
+    MissingJobRecordError,
+    get_job_record,
+    to_job_ids,
+    update_job_records,
+)
+from covalent_dispatcher._db.models import Job, Lattice
+from covalent_dispatcher._db.write_result_to_db import insert_electrons_data, insert_lattices_data
+
+from .write_result_to_db_test import get_electron_kwargs, get_lattice_kwargs
+
+
+@pytest.fixture
+def test_db():
+    """Instantiate and return an in-memory database"""
+    return DataStore(
+        db_URL="sqlite+pysqlite:///:memory:",
+        initialize_db=True,
+    )
+
+
+def test_get_job_record(test_db, mocker):
+    """
+    Test retrieving a job record from database
+    """
+    mocker.patch("covalent_dispatcher._db.jobdb.workflow_db", test_db)
+    with test_db.session() as session:
+        job = Job(cancel_requested=True, job_handle="aws_job_id")
+        session.add(job)
+
+    record = get_job_record(1)
+    assert record["job_handle"] == "aws_job_id"
+
+    with pytest.raises(MissingJobRecordError):
+        get_job_record(42)
+
+
+def test_update_job_records(test_db, mocker):
+    """
+    Test updating job records
+    """
+    mocker.patch("covalent_dispatcher._db.jobdb.workflow_db", test_db)
+    with test_db.session() as session:
+        job = Job(cancel_requested=False, job_handle="aws_job_id_1")
+        session.add(job)
+
+        job = Job(cancel_requested=False, job_handle="aws_job_id_2")
+        session.add(job)
+
+    job_1_kwargs = {"job_id": 1, "cancel_requested": True, "cancel_successful": True}
+    job_2_kwargs = {"job_id": 2, "cancel_requested": True, "job_handle": "42"}
+
+    update_job_records([job_1_kwargs, job_2_kwargs])
+
+    record_1 = get_job_record(1)
+    record_2 = get_job_record(2)
+    assert record_1["cancel_requested"] is True
+    assert record_1["cancel_successful"] is True
+    assert record_2["cancel_requested"] is True
+    assert record_2["cancel_successful"] is False
+
+    with pytest.raises(MissingJobRecordError):
+        update_job_records([{"job_id": 5, "cancel_requested": True}])
+
+
+def test_to_job_ids(test_db, mocker):
+    """
+    Test mapping task ids to job ids
+    """
+    mocker.patch("covalent_dispatcher._db.jobdb.workflow_db", test_db)
+    mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
+    cur_time = dt.now(timezone.utc)
+    lattice_kwargs = get_lattice_kwargs(
+        dispatch_id="test_dispatch", created_at=cur_time, updated_at=cur_time, started_at=cur_time
+    )
+    insert_lattices_data(**lattice_kwargs)
+
+    with test_db.session() as session:
+        rows = session.query(Lattice).all()
+        assert len(rows) == 1
+
+    electron_kwargs = {
+        **get_electron_kwargs(
+            parent_dispatch_id="test_dispatch",
+            transport_graph_node_id=0,
+            cancel_requested=False,
+            created_at=cur_time,
+            updated_at=cur_time,
+        )
+    }
+
+    insert_electrons_data(**electron_kwargs)
+
+    electron_kwargs = {
+        **get_electron_kwargs(
+            parent_dispatch_id="test_dispatch",
+            transport_graph_node_id=1,
+            cancel_requested=False,
+            created_at=cur_time,
+            updated_at=cur_time,
+        )
+    }
+
+    insert_electrons_data(**electron_kwargs)
+
+    job_ids = to_job_ids("test_dispatch", [0])
+    assert job_ids == [1]
+
+    job_ids = to_job_ids("test_dispatch", [0, 1])
+    assert job_ids == [1, 2]
diff --git a/tests/covalent_dispatcher_tests/_db/update_test.py b/tests/covalent_dispatcher_tests/_db/update_test.py
index 44a6882c..7db17d8a 100644
--- a/tests/covalent_dispatcher_tests/_db/update_test.py
+++ b/tests/covalent_dispatcher_tests/_db/update_test.py
@@ -32,7 +32,7 @@ from covalent._workflow.lattice import Lattice as LatticeClass
 from covalent.executor import LocalExecutor
 from covalent_dispatcher._db import update, upsert
 from covalent_dispatcher._db.datastore import DataStore
-from covalent_dispatcher._db.models import Electron, ElectronDependency, Lattice
+from covalent_dispatcher._db.models import Electron, ElectronDependency, Job, Lattice
 from covalent_dispatcher._db.write_result_to_db import load_file
 from covalent_dispatcher._service.app import _result_from
 
@@ -73,6 +73,25 @@ def result_1():
     return result
 
 
+@pytest.fixture
+def result_2():
+    @ct.electron(executor="dask")
+    def task(x):
+        return x
+
+    @ct.lattice(executor="local", workflow_executor="local")
+    def workflow_2(x):
+        return x
+
+    Path(f"{TEMP_RESULTS_DIR}/dispatch_1").mkdir(parents=True, exist_ok=True)
+    workflow_2.build_graph(x=2)
+    received_lattice = LatticeClass.deserialize_from_json(workflow_2.serialize_to_json())
+    result = Result(received_lattice, dispatch_id="dispatch_2")
+
+    result._initialize_nodes()
+    return result
+
+
 @pytest.fixture
 def test_db():
     """Instantiate and return an in-memory database."""
@@ -85,7 +104,6 @@ def test_db():
 
 def test_update_node(test_db, result_1, mocker):
     """Test the node update method."""
-
     mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
     mocker.patch("covalent_dispatcher._db.upsert.workflow_db", test_db)
     update.persist(result_1)
@@ -312,31 +330,43 @@ def test_result_persist_workflow_1(test_db, result_1, mocker):
     teardown_temp_results_dir(dispatch_id="dispatch_1")
 
 
-def test_result_persist_subworkflow_1(test_db, result_1, mocker):
+def test_result_persist_subworkflow_1(test_db, result_1, result_2, mocker):
     """Test the persist method for the Result object when passed an electron_id"""
 
     mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
     mocker.patch("covalent_dispatcher._db.upsert.workflow_db", test_db)
-    update.persist(result_1, electron_id=2)
+    update.persist(result_1)
+    update.persist(result_2, electron_id=1)
+
+    with test_db.session() as session:
+        electron = session.query(Electron).where(Electron.id == 1).first()
+        job_record = session.query(Job).where(Job.id == Electron.job_id).first()
+        job_record.cancel_requested = True
 
     # Query lattice / electron / electron dependency
     with test_db.session() as session:
-        lattice_row = session.query(Lattice).first()
-        electron_rows = session.query(Electron).all()
-        electron_dependency_rows = session.query(ElectronDependency).all()
+        lattice_row = session.query(Lattice).where(Lattice.dispatch_id == "dispatch_2").first()
+        job_records = (
+            session.query(Electron, Job)
+            .where(Job.id == Electron.job_id, Electron.parent_lattice_id == lattice_row.id)
+            .all()
+        )
 
         # Check that lattice record is as expected
-        assert lattice_row.dispatch_id == "dispatch_1"
+        assert lattice_row.dispatch_id == "dispatch_2"
         assert isinstance(lattice_row.created_at, dt)
         assert lattice_row.started_at is None
         assert isinstance(lattice_row.updated_at, dt) and isinstance(lattice_row.created_at, dt)
         assert lattice_row.completed_at is None
         assert lattice_row.status == "NEW_OBJECT"
-        assert lattice_row.name == "workflow_1"
-        assert lattice_row.electron_id == 2
+        assert lattice_row.name == "workflow_2"
+        assert lattice_row.electron_id == 1
         assert lattice_row.executor == "local"
         assert lattice_row.workflow_executor == "local"
 
+        for job in job_records:
+            assert job.cancel_requested is True
+
 
 def test_result_persist_rehydrate(test_db, result_1, mocker):
     """Test that persist followed by result_from preserves all result,
diff --git a/tests/covalent_dispatcher_tests/_db/upsert_test.py b/tests/covalent_dispatcher_tests/_db/upsert_test.py
index e82e10f7..bd1159ee 100644
--- a/tests/covalent_dispatcher_tests/_db/upsert_test.py
+++ b/tests/covalent_dispatcher_tests/_db/upsert_test.py
@@ -32,7 +32,9 @@ from covalent_dispatcher._db.upsert import (
     ELECTRON_RESULTS_FILENAME,
     ELECTRON_STDERR_FILENAME,
     ELECTRON_STDOUT_FILENAME,
-    _electron_data,
+    LATTICE_FUNCTION_STRING_FILENAME,
+    electron_data,
+    lattice_data,
 )
 
 TEMP_RESULTS_DIR = os.environ.get("COVALENT_DATA_DIR") or ct.get_config("dispatcher.results_dir")
@@ -80,7 +82,7 @@ def test_upsert_electron_data_handles_missing_keys(test_db, result_1, mocker):
     mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
     mocker.patch("covalent_dispatcher._db.upsert.workflow_db", test_db)
     mock_store_file = mocker.patch("covalent_dispatcher._db.upsert.store_file")
-    mocker.patch("covalent_dispatcher._db.upsert.insert_electrons_data")
+    mocker.patch("covalent_dispatcher._db.upsert.transaction_insert_electrons_data")
     mocker.patch("covalent_dispatcher._db.write_result_to_db.update_electrons_data")
     mocker.patch(
         "covalent_dispatcher._db.write_result_to_db.update_lattice_completed_electron_num"
@@ -92,10 +94,30 @@ def test_upsert_electron_data_handles_missing_keys(test_db, result_1, mocker):
     del tg._graph.nodes[0]["stdout"]
     del tg._graph.nodes[0]["output"]
 
-    _electron_data(result_1)
+    electron_data(result_1)
 
     node_path = Path(TEMP_RESULTS_DIR) / result_1.dispatch_id / "node_0"
     mock_store_file.assert_any_call(node_path, ELECTRON_ERROR_FILENAME, None)
     mock_store_file.assert_any_call(node_path, ELECTRON_STDOUT_FILENAME, None)
     mock_store_file.assert_any_call(node_path, ELECTRON_STDERR_FILENAME, None)
     mock_store_file.assert_any_call(node_path, ELECTRON_RESULTS_FILENAME, None)
+
+
+def test_public_lattice_data(test_db, result_1, mocker):
+    """Test the lattice data public method"""
+    mocker.patch("covalent_dispatcher._db.upsert.workflow_db", test_db)
+    mock_store_file = mocker.patch("covalent_dispatcher._db.upsert.store_file")
+    mock_insert = mocker.patch("covalent_dispatcher._db.upsert.transaction_insert_lattices_data")
+    mocker.patch("covalent_dispatcher._db.upsert.transaction_update_lattices_data")
+
+    lattice_path = str(Path(TEMP_RESULTS_DIR) / result_1.dispatch_id)
+
+    lattice_data(result_1)
+    mock_store_file.assert_any_call(
+        lattice_path, LATTICE_FUNCTION_STRING_FILENAME, result_1.lattice.workflow_function_string
+    )
+
+    del result_1.lattice.__dict__["workflow_function_string"]
+    mock_store_file.reset_mock()
+    lattice_data(result_1)
+    mock_store_file.assert_any_call(lattice_path, LATTICE_FUNCTION_STRING_FILENAME, None)
diff --git a/tests/covalent_dispatcher_tests/_db/write_result_to_db_test.py b/tests/covalent_dispatcher_tests/_db/write_result_to_db_test.py
index c8c05c0f..6edb6e23 100644
--- a/tests/covalent_dispatcher_tests/_db/write_result_to_db_test.py
+++ b/tests/covalent_dispatcher_tests/_db/write_result_to_db_test.py
@@ -39,7 +39,7 @@ from covalent._shared_files.defaults import (
     subscript_prefix,
 )
 from covalent_dispatcher._db.datastore import DataStore
-from covalent_dispatcher._db.models import Electron, ElectronDependency, Lattice
+from covalent_dispatcher._db.models import Electron, ElectronDependency, Job, Lattice
 from covalent_dispatcher._db.write_result_to_db import (
     InvalidFileExtension,
     MissingElectronRecordError,
@@ -52,10 +52,10 @@ from covalent_dispatcher._db.write_result_to_db import (
     load_file,
     resolve_electron_id,
     store_file,
+    transaction_upsert_electron_dependency_data,
     update_electrons_data,
     update_lattice_completed_electron_num,
     update_lattices_data,
-    upsert_electron_dependency_data,
 )
 
 STORAGE_TYPE = "local"
@@ -209,6 +209,7 @@ def get_electron_kwargs(
     deps_filename=DEPS_FILENAME,
     call_before_filename=CALL_BEFORE_FILENAME,
     call_after_filename=CALL_AFTER_FILENAME,
+    cancel_requested=False,
     created_at=None,
     updated_at=None,
     started_at=None,
@@ -236,6 +237,7 @@ def get_electron_kwargs(
         "deps_filename": deps_filename,
         "call_before_filename": call_before_filename,
         "call_after_filename": call_after_filename,
+        "cancel_requested": cancel_requested,
         "created_at": created_at,
         "updated_at": updated_at,
         "started_at": started_at,
@@ -328,7 +330,8 @@ def test_insert_lattices_data(test_db, mocker):
         assert not rows
 
 
-def test_insert_electrons_data(test_db, mocker):
+@pytest.mark.parametrize("cancel_requested", [True, False])
+def test_insert_electrons_data(cancel_requested, test_db, mocker):
     """Test the function that inserts the electron data to the Electrons table."""
 
     mocker.patch("covalent_dispatcher._db.write_result_to_db.workflow_db", test_db)
@@ -339,6 +342,7 @@ def test_insert_electrons_data(test_db, mocker):
 
     electron_kwargs = {
         **get_electron_kwargs(
+            cancel_requested=cancel_requested,
             created_at=cur_time,
             updated_at=cur_time,
         )
@@ -358,11 +362,15 @@ def test_insert_electrons_data(test_db, mocker):
                     assert getattr(electron, key).strftime("%m/%d/%Y, %H:%M:%S") == value.strftime(
                         "%m/%d/%Y, %H:%M:%S"
                     )
+                elif key == "cancel_requested":
+                    continue
                 else:
                     assert getattr(electron, key) == value
             assert electron.is_active
 
-        insert_electrons_data(**electron_kwargs)
+        rows = session.query(Job).all()
+        assert len(rows) == 1
+        assert rows[0].cancel_requested == cancel_requested
 
 
 def test_insert_electrons_data_missing_lattice_record(test_db, mocker):
@@ -476,12 +484,17 @@ def test_upsert_electron_dependency_data(test_db, workflow_lattice, mocker):
         electron_ids.append(insert_electrons_data(**electron_kwargs))
 
     mock_insert = mocker.patch(
-        "covalent_dispatcher._db.write_result_to_db.insert_electron_dependency_data"
+        "covalent_dispatcher._db.write_result_to_db.transaction_insert_electron_dependency_data"
     )
 
-    upsert_electron_dependency_data(dispatch_id="dispatch_1", lattice=workflow_lattice)
+    with test_db.session() as session:
+        transaction_upsert_electron_dependency_data(
+            session=session, dispatch_id="dispatch_1", lattice=workflow_lattice
+        )
 
-    mock_insert.assert_called_once_with(dispatch_id="dispatch_1", lattice=workflow_lattice)
+        mock_insert.assert_called_once_with(
+            session=session, dispatch_id="dispatch_1", lattice=workflow_lattice
+        )
 
 
 def test_upsert_electron_dependency_data_idempotent(test_db, workflow_lattice, mocker):
@@ -515,7 +528,12 @@ def test_upsert_electron_dependency_data_idempotent(test_db, workflow_lattice, m
     mock_insert = mocker.patch(
         "covalent_dispatcher._db.write_result_to_db.insert_electron_dependency_data"
     )
-    upsert_electron_dependency_data(dispatch_id="dispatch_1", lattice=workflow_lattice)
+
+    with test_db.session() as session:
+        transaction_upsert_electron_dependency_data(
+            session=session, dispatch_id="dispatch_1", lattice=workflow_lattice
+        )
+
     mock_insert.assert_not_called()
 
 
@@ -583,7 +601,10 @@ def test_update_electrons_data(test_db, mocker):
         )
 
     insert_electrons_data(
-        **get_electron_kwargs(created_at=dt.now(timezone.utc), updated_at=dt.now(timezone.utc)),
+        **get_electron_kwargs(
+            created_at=dt.now(timezone.utc),
+            updated_at=dt.now(timezone.utc),
+        ),
     )
     cur_time = dt.now(timezone.utc)
     update_electrons_data(
diff --git a/tests/covalent_dispatcher_tests/_service/app_test.py b/tests/covalent_dispatcher_tests/_service/app_test.py
index 0a9b88c0..36133836 100644
--- a/tests/covalent_dispatcher_tests/_service/app_test.py
+++ b/tests/covalent_dispatcher_tests/_service/app_test.py
@@ -61,6 +61,11 @@ class MockDataStore:
             yield session
 
 
+@pytest.fixture
+def app():
+    yield fast_app
+
+
 @pytest.fixture
 def client():
     with TestClient(fast_app) as c:
@@ -126,6 +131,28 @@ async def test_redispatch(mocker, client):
     )
 
 
+def test_cancel_dispatch(mocker, app, client):
+    """
+    Test cancelling dispatch
+    """
+    mocker.patch("covalent_dispatcher.cancel_running_dispatch")
+    response = client.post(
+        "/api/cancel", data=json.dumps({"dispatch_id": DISPATCH_ID, "task_ids": []})
+    )
+    assert response.json() == f"Dispatch {DISPATCH_ID} cancelled."
+
+
+def test_cancel_tasks(mocker, app, client):
+    """
+    Test cancelling tasks within a lattice after dispatch
+    """
+    mocker.patch("covalent_dispatcher.cancel_running_dispatch")
+    response = client.post(
+        "/api/cancel", data=json.dumps({"dispatch_id": DISPATCH_ID, "task_ids": [0, 1]})
+    )
+    assert response.json() == f"Cancelled tasks [0, 1] in dispatch {DISPATCH_ID}."
+
+
 @pytest.mark.asyncio
 async def test_redispatch_exception(mocker, client):
     """Test the redispatch endpoint."""
@@ -143,9 +170,11 @@ async def test_cancel(mocker, client):
     cancel_running_dispatch_mock = mocker.patch(
         "covalent_dispatcher.cancel_running_dispatch", return_value=DISPATCH_ID
     )
-    response = client.post("/api/cancel", data=DISPATCH_ID.encode("utf-8"))
+    response = client.post(
+        "/api/cancel", data=json.dumps({"dispatch_id": DISPATCH_ID, "task_ids": []})
+    )
     assert response.json() == f"Dispatch {DISPATCH_ID} cancelled."
-    cancel_running_dispatch_mock.assert_called_once_with(DISPATCH_ID)
+    cancel_running_dispatch_mock.assert_called_once_with(DISPATCH_ID, [])
 
 
 @pytest.mark.asyncio
@@ -154,10 +183,14 @@ async def test_cancel_exception(mocker, client):
     cancel_running_dispatch_mock = mocker.patch(
         "covalent_dispatcher.cancel_running_dispatch", side_effect=Exception("mock")
     )
-    response = client.post("/api/cancel", data=DISPATCH_ID.encode("utf-8"))
-    assert response.status_code == 400
-    assert response.json()["detail"] == "Failed to cancel workflow: mock"
-    cancel_running_dispatch_mock.assert_called_once_with(DISPATCH_ID)
+
+    with pytest.raises(Exception):
+        response = client.post(
+            "/api/cancel", data=json.dumps({"dispatch_id": DISPATCH_ID, "task_ids": []})
+        )
+        assert response.status_code == 400
+        assert response.json()["detail"] == "Failed to cancel workflow: mock"
+        cancel_running_dispatch_mock.assert_called_once_with(DISPATCH_ID, [])
 
 
 def test_get_result(mocker, client, test_db_file):
diff --git a/tests/covalent_dispatcher_tests/entry_point_test.py b/tests/covalent_dispatcher_tests/entry_point_test.py
index 3a9a3c06..a5d1a72c 100644
--- a/tests/covalent_dispatcher_tests/entry_point_test.py
+++ b/tests/covalent_dispatcher_tests/entry_point_test.py
@@ -63,7 +63,8 @@ async def test_run_redispatch(mocker):
     make_derived_dispatch_mock.assert_called_once_with(DISPATCH_ID, "mock-json-lattice", {}, False)
 
 
-def test_cancel_running_dispatch(mocker):
-    mock_cancel_workflow = mocker.patch("covalent_dispatcher._core.cancel_workflow")
-    cancel_running_dispatch(DISPATCH_ID)
-    mock_cancel_workflow.assert_called_once_with(DISPATCH_ID)
+@pytest.mark.asyncio
+async def test_cancel_running_dispatch(mocker):
+    mock_cancel_workflow = mocker.patch("covalent_dispatcher.entry_point.cancel_dispatch")
+    await cancel_running_dispatch(DISPATCH_ID)
+    mock_cancel_workflow.assert_awaited_once_with(DISPATCH_ID, [])
diff --git a/tests/covalent_tests/executor/base_test.py b/tests/covalent_tests/executor/base_test.py
index 7b38d0ac..0aebc0a4 100644
--- a/tests/covalent_tests/executor/base_test.py
+++ b/tests/covalent_tests/executor/base_test.py
@@ -28,9 +28,11 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 from covalent import DepsCall, TransportableObject
-from covalent._shared_files.exceptions import TaskRuntimeError
+from covalent._results_manager import Result
+from covalent._shared_files.exceptions import TaskCancelledError, TaskRuntimeError
 from covalent.executor import BaseExecutor, wrapper_fn
 from covalent.executor.base import AsyncBaseExecutor
+from covalent.executor.utils.wrappers import Signals
 
 
 class MockExecutor(BaseExecutor):
@@ -219,6 +221,7 @@ def test_base_executor_execute(mocker):
     node_id = -1
 
     assembled_callable = partial(wrapper_fn, function, call_before, call_after)
+    mock_notify = mocker.patch("covalent.executor.BaseExecutor._notify")
 
     result, stdout, stderr, exception_raised = me.execute(
         function=assembled_callable,
@@ -230,6 +233,7 @@ def test_base_executor_execute(mocker):
     )
 
     assert result.get_deserialized() == 5
+    mock_notify.assert_called_with(Signals.EXIT)
 
 
 @pytest.mark.asyncio
@@ -343,6 +347,7 @@ def test_base_executor_passes_task_metadata(mocker):
     node_id = -1
 
     assembled_callable = partial(wrapper_fn, function, call_before, call_after)
+    mock_notify = mocker.patch("covalent.executor.BaseExecutor._notify")
 
     metadata, stdout, stderr, exception_raised = me.execute(
         function=assembled_callable,
@@ -354,6 +359,7 @@ def test_base_executor_passes_task_metadata(mocker):
     )
     task_metadata = {"dispatch_id": dispatch_id, "node_id": node_id, "results_dir": results_dir}
     assert metadata == task_metadata
+    mock_notify.assert_called_with(Signals.EXIT)
 
 
 def test_base_async_executor_passes_task_metadata(mocker):
@@ -377,6 +383,7 @@ def test_base_async_executor_passes_task_metadata(mocker):
     node_id = -1
 
     assembled_callable = partial(wrapper_fn, function, call_before, call_after)
+    mock_notify = mocker.patch("covalent.executor.base.AsyncBaseExecutor._notify")
 
     awaitable = me.execute(
         function=assembled_callable,
@@ -390,6 +397,7 @@ def test_base_async_executor_passes_task_metadata(mocker):
     metadata, stdout, stderr, exception_raised = asyncio.run(awaitable)
     task_metadata = {"dispatch_id": dispatch_id, "node_id": node_id, "results_dir": results_dir}
     assert metadata == task_metadata
+    mock_notify.assert_called_with(Signals.EXIT)
 
 
 def test_async_write_streams_to_file(mocker):
@@ -454,6 +462,7 @@ def test_executor_setup_teardown_method(mocker):
     }
 
     assembled_callable = partial(wrapper_fn, function, call_before, call_after)
+    mock_notify = mocker.patch("covalent.executor.BaseExecutor._notify")
 
     result, stdout, stderr, exception_raised = me.execute(
         function=assembled_callable,
@@ -467,6 +476,7 @@ def test_executor_setup_teardown_method(mocker):
     assert result.get_deserialized() == 5
     me.setup.assert_called_once_with(task_metadata=task_metadata)
     me.teardown.assert_called_once_with(task_metadata=task_metadata)
+    mock_notify.assert_called_with(Signals.EXIT)
 
 
 def test_async_executor_setup_teardown(mocker):
@@ -489,6 +499,7 @@ def test_async_executor_setup_teardown(mocker):
     node_id = -1
 
     assembled_callable = partial(wrapper_fn, function, call_before, call_after)
+    mock_notify = mocker.patch("covalent.executor.base.AsyncBaseExecutor._notify")
 
     awaitable = me.execute(
         function=assembled_callable,
@@ -504,6 +515,7 @@ def test_async_executor_setup_teardown(mocker):
     me.run.assert_called_once_with(assembled_callable, args, kwargs, task_metadata)
     me.setup.assert_called_once_with(task_metadata=task_metadata)
     me.teardown.assert_called_once_with(task_metadata=task_metadata)
+    mock_notify.assert_called_with(Signals.EXIT)
 
 
 def test_executor_from_dict_makes_deepcopy():
@@ -525,7 +537,7 @@ def test_executor_execute_runtime_error_handling(mocker):
         return x, y
 
     me = MockExecutor(log_stdout="/tmp/stdout.log")
-    me.run = MagicMock(side_effect=TaskRuntimeError("error"))
+    me.run = MagicMock(side_effect=TaskCancelledError("error"))
 
     function = TransportableObject(f)
     args = [TransportableObject(2)]
@@ -537,8 +549,8 @@ def test_executor_execute_runtime_error_handling(mocker):
     node_id = -1
 
     assembled_callable = partial(wrapper_fn, function, call_before, call_after)
-
-    output, stdout, stderr, exception_raised = me.execute(
+    mock_notify = mocker.patch("covalent.executor.BaseExecutor._notify")
+    output, stdout, stderr, job_status = me.execute(
         function=assembled_callable,
         args=args,
         kwargs=kwargs,
@@ -547,7 +559,8 @@ def test_executor_execute_runtime_error_handling(mocker):
         node_id=node_id,
     )
 
-    assert exception_raised is True
+    assert job_status is Result.CANCELLED
+    mock_notify.assert_called_with(Signals.EXIT)
 
 
 @pytest.mark.asyncio
@@ -558,7 +571,112 @@ async def test_async_base_executor_execute_runtime_error_handling(mocker):
         return x, y
 
     me = MockAsyncExecutor(log_stdout="/tmp/stdout.log")
-    me.run = AsyncMock(side_effect=TaskRuntimeError("error"))
+    me.run = AsyncMock(side_effect=TaskCancelledError("error"))
+
+    function = TransportableObject(f)
+    args = [TransportableObject(2)]
+    kwargs = {"y": TransportableObject(3)}
+    call_before = []
+    call_after = []
+    dispatch_id = "asdf"
+    results_dir = "/tmp"
+    node_id = -1
+
+    assembled_callable = partial(wrapper_fn, function, call_before, call_after)
+    mock_notify = mocker.patch("covalent.executor.base.AsyncBaseExecutor._notify")
+
+    output, stdout, stderr, job_status = await me.execute(
+        function=assembled_callable,
+        args=args,
+        kwargs=kwargs,
+        dispatch_id=dispatch_id,
+        results_dir=results_dir,
+        node_id=node_id,
+    )
+
+    assert job_status is Result.CANCELLED
+    mock_notify.assert_called_with(Signals.EXIT)
+
+
+def test_base_executor_get_cancel_requested(mocker):
+    """
+    Test executor invoking get cancel requested
+    """
+    me = MockExecutor()
+    me._init_runtime()
+    recv_queue = me._recv_queue
+    recv_queue.put_nowait((True, True))
+    mock_notify = mocker.patch("covalent.executor.base.BaseExecutor._notify")
+
+    assert me.get_cancel_requested() is True
+    mock_notify.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_async_base_executor_get_cancel_requested(mocker):
+    me = MockAsyncExecutor()
+    me._init_runtime()
+    send_queue = me._send_queue
+    recv_queue = me._recv_queue
+    recv_queue.put_nowait((True, True))
+    assert await me.get_cancel_requested() is True
+
+
+def test_base_executor_set_job_handle(mocker):
+    me = MockExecutor()
+    me._init_runtime()
+    send_queue = me._send_queue
+    recv_queue = me._recv_queue
+    recv_queue.put_nowait((True, None))
+
+    mock_notify = mocker.patch("covalent.executor.base.BaseExecutor._notify")
+    me.set_job_handle(42)
+
+    mock_notify.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_async_base_executor_set_job_handle(mocker):
+    me = MockAsyncExecutor()
+    me._init_runtime()
+    send_queue = me._send_queue
+    recv_queue = me._recv_queue
+
+    mock_notify = mocker.patch("covalent.executor.base.AsyncBaseExecutor._notify")
+    recv_queue.put_nowait((True, None))
+    await me.set_job_handle(42)
+
+    mock_notify.assert_called_once()
+
+
+def test_base_executor_notify(mocker):
+    me = MockExecutor()
+    me._init_runtime(loop=MagicMock())
+    me._loop.call_soon_threadsafe = MagicMock()
+
+    me._notify("get")
+    me._loop.call_soon_threadsafe.assert_called_with(me._send_queue.put_nowait, ("get", None))
+
+
+def test_wait_for_response_runtimeerror(mocker):
+    me = MockExecutor()
+    me._init_runtime()
+    me._recv_queue.get = MagicMock(return_value=(None, None))
+
+    with pytest.raises(RuntimeError):
+        me._wait_for_response()
+
+    me._recv_queue.get.assert_called()
+
+
+def test_base_executor_executor_task_runtime_error(mocker):
+    """Check handling of `TaskRuntimeError` exceptions"""
+
+    def f(x, y):
+        return x, y
+
+    me = MockExecutor(log_stdout="/tmp/stdout.log")
+    me.run = MagicMock(side_effect=TaskRuntimeError("error"))
 
     function = TransportableObject(f)
     args = [TransportableObject(2)]
@@ -570,8 +688,85 @@ async def test_async_base_executor_execute_runtime_error_handling(mocker):
     node_id = -1
 
     assembled_callable = partial(wrapper_fn, function, call_before, call_after)
+    mock_notify = mocker.patch("covalent.executor.BaseExecutor._notify")
+    output, stdout, stderr, job_status = me.execute(
+        function=assembled_callable,
+        args=args,
+        kwargs=kwargs,
+        dispatch_id=dispatch_id,
+        results_dir=results_dir,
+        node_id=node_id,
+    )
+
+    assert job_status is Result.FAILED
+    mock_notify.assert_called_with(Signals.EXIT)
+
+
+def test_base_executor_cancel_app_log(mocker):
+    me = MockExecutor()
+    me._init_runtime()
+    mock_app_log = mocker.patch("covalent.executor.base.app_log.debug")
+    task_metadata = {}
+    job_handle = 42
+
+    result = me.cancel(task_metadata, job_handle)
+    mock_app_log.assert_called_once()
+    assert result is False
+
+
+@pytest.mark.asyncio
+async def test_base_executor_cancel(mocker):
+    me = MockExecutor()
+    me._init_runtime(loop=AsyncMock(), cancel_pool=MagicMock())
+    me._loop.run_in_executor = AsyncMock()
+    await me._cancel({}, 42)
+    me._loop.run_in_executor.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_base_async_executor_wait_for_response_raises_runtimeerror(mocker):
+    me = MockAsyncExecutor()
+    me._init_runtime()
+    me._recv_queue = AsyncMock()
+
+    mock_async_wait_for = mocker.patch(
+        "covalent.executor.base.asyncio.wait_for",
+        return_value=(
+            False,
+            None,
+        ),
+    )
+
+    with pytest.raises(RuntimeError):
+        await me._wait_for_response()
+        me._recv_queue.get.assert_called()
+        mock_async_wait_for.assert_awaited()
+
+
+@pytest.mark.asyncio
+async def test_base_async_executor_task_runtime_error(mocker):
+    me = MockAsyncExecutor()
+    me._init_runtime()
+
+    def f(x, y):
+        return x, y
 
-    output, stdout, stderr, exception_raised = await me.execute(
+    me.setup = AsyncMock()
+    me.teardown = AsyncMock()
+    me._notify = MagicMock()
+    me.run = AsyncMock(side_effect=TaskRuntimeError("error"))
+    function = TransportableObject(f)
+    args = [TransportableObject(2)]
+    kwargs = {"y": TransportableObject(3)}
+    call_before = []
+    call_after = []
+    dispatch_id = "asdf"
+    results_dir = "/tmp"
+    node_id = -1
+    task_metadata = {"dispatch_id": dispatch_id, "node_id": node_id, "results_dir": results_dir}
+
+    assembled_callable = partial(wrapper_fn, function, call_before, call_after)
+    output, stdout, stderr, job_status = await me.execute(
         function=assembled_callable,
         args=args,
         kwargs=kwargs,
@@ -580,4 +775,33 @@ async def test_async_base_executor_execute_runtime_error_handling(mocker):
         node_id=node_id,
     )
 
-    assert exception_raised is True
+    me.setup.assert_awaited()
+    me.run.assert_awaited_once_with(assembled_callable, args, kwargs, task_metadata)
+    assert job_status is Result.FAILED
+    me._notify.assert_called_with(Signals.EXIT)
+
+
+@pytest.mark.asyncio
+async def test_base_async_executor_task_cancel(mocker):
+    mock_app_log = mocker.patch("covalent.executor.base.app_log.debug")
+    me = MockAsyncExecutor()
+    task_metadata = {}
+    job_handle = 42
+    result = await me.cancel(task_metadata=task_metadata, job_handle=job_handle)
+    mock_app_log.assert_called_with(f"Cancel not implemented for executor {type(me)}")
+    assert result is False
+
+
+@pytest.mark.asyncio
+async def test_base_async_executor_private_cancel(mocker):
+    mock_app_log = mocker.patch("covalent.executor.base.app_log.debug")
+    me = MockAsyncExecutor()
+    me.teardown = AsyncMock()
+
+    task_metadata = {}
+    job_handle = 42
+
+    cancel_result = await me._cancel(task_metadata=task_metadata, job_handle=job_handle)
+    mock_app_log.assert_called_with(f"Cancel not implemented for executor {type(me)}")
+    me.teardown.assert_awaited_with(task_metadata)
+    assert cancel_result is False
diff --git a/tests/covalent_tests/executor/executor_plugins/dask_test.py b/tests/covalent_tests/executor/executor_plugins/dask_test.py
index ba3c0db3..9f72a817 100644
--- a/tests/covalent_tests/executor/executor_plugins/dask_test.py
+++ b/tests/covalent_tests/executor/executor_plugins/dask_test.py
@@ -21,10 +21,13 @@
 """Tests for Covalent dask executor."""
 
 import asyncio
+from unittest.mock import AsyncMock
 
 import pytest
 
 from covalent._shared_files import TaskRuntimeError
+from covalent._shared_files.exceptions import TaskCancelledError
+from covalent.executor.executor_plugins.dask import DaskExecutor
 
 
 def test_dask_executor_init(mocker):
@@ -74,7 +77,7 @@ def test_dask_wrapper_fn_exception_handling(mocker):
     assert output is None
 
 
-def test_dask_executor_run():
+def test_dask_executor_run(mocker):
     """Test run method for Dask executor"""
 
     import io
@@ -87,6 +90,10 @@ def test_dask_executor_run():
     cluster = LocalCluster()
 
     dask_exec = DaskExecutor(cluster.scheduler_address)
+    mock_get_cancel_requested = mocker.patch.object(
+        dask_exec, "get_cancel_requested", AsyncMock(return_value=False)
+    )
+    mock_set_job_handle = mocker.patch.object(dask_exec, "set_job_handle", AsyncMock())
 
     def f(x, y):
         print("Hello", file=sys.stdout)
@@ -103,9 +110,46 @@ def test_dask_executor_run():
     assert result == (5, 7)
     assert dask_exec.task_stdout.getvalue() == "Hello\n"
     assert dask_exec.task_stderr.getvalue() == "Bye\n"
+    mock_get_cancel_requested.assert_awaited()
+    mock_set_job_handle.assert_awaited()
 
 
-def test_dask_executor_run_exception_handling():
+def test_dask_executor_run_cancel_requested(mocker):
+    """
+    Test dask executor cancel request
+    """
+    import io
+    import sys
+
+    from dask.distributed import LocalCluster
+
+    from covalent.executor import DaskExecutor
+
+    cluster = LocalCluster()
+
+    dask_exec = DaskExecutor(cluster.scheduler_address)
+    mock_get_cancel_requested = mocker.patch.object(
+        dask_exec, "get_cancel_requested", AsyncMock(side_effect=TaskCancelledError)
+    )
+    mock_set_job_handle = mocker.patch.object(dask_exec, "set_job_handle", AsyncMock())
+
+    def f(x, y):
+        print("Hello", file=sys.stdout)
+        print("Bye", file=sys.stderr)
+        return x, y
+
+    args = [5]
+    kwargs = {"y": 7}
+    task_metadata = {"dispatch_id": "asdf", "node_id": 1}
+    dask_exec._task_stdout = io.StringIO()
+    dask_exec._task_stderr = io.StringIO()
+    with pytest.raises(TaskCancelledError):
+        asyncio.run(dask_exec.run(f, args, kwargs, task_metadata))
+        mock_get_cancel_requested.assert_awaited()
+        mock_set_job_handle.assert_awaited()
+
+
+def test_dask_executor_run_exception_handling(mocker):
     """Test run method for Dask executor"""
 
     import io
@@ -118,6 +162,10 @@ def test_dask_executor_run_exception_handling():
     cluster = LocalCluster()
 
     dask_exec = DaskExecutor(cluster.scheduler_address)
+    mock_get_cancel_requested = mocker.patch.object(
+        dask_exec, "get_cancel_requested", AsyncMock(return_value=False)
+    )
+    mock_set_job_handle = mocker.patch.object(dask_exec, "set_job_handle", AsyncMock())
     dask_exec._task_stdout = io.StringIO()
     dask_exec._task_stderr = io.StringIO()
 
@@ -133,3 +181,60 @@ def test_dask_executor_run_exception_handling():
 
     dask_exec._task_stdout.getvalue() == "f output"
     assert "RuntimeError" in dask_exec._task_stderr.getvalue()
+    mock_get_cancel_requested.assert_awaited()
+    mock_set_job_handle.assert_awaited()
+
+
+def test_dask_app_log_debug_when_cancel_requested(mocker):
+    """
+    Test logging when task cancellation is requested
+    """
+    from dask.distributed import LocalCluster
+
+    from covalent.executor import DaskExecutor
+
+    cluster = LocalCluster()
+
+    mock_app_log = mocker.patch("covalent.executor.executor_plugins.dask.app_log.debug")
+
+    dask_exec = DaskExecutor(cluster.scheduler_address)
+    mock_get_cancel_requested = mocker.patch.object(
+        dask_exec, "get_cancel_requested", AsyncMock(return_value=True)
+    )
+
+    def f(x, y):
+        print("f output")
+        raise RuntimeError("error")
+
+    args = [5]
+    kwargs = {"y": 7}
+    task_metadata = {"dispatch_id": "asdf", "node_id": 1}
+    try:
+        asyncio.run(dask_exec.run(f, args, kwargs, task_metadata))
+    except TaskCancelledError:
+        pass
+    finally:
+        mock_get_cancel_requested.assert_awaited_once()
+        mock_app_log.assert_called_once()
+
+
+def test_dask_task_cancel(mocker):
+    """
+    Test dask task cancellation method
+    """
+    from dask.distributed import LocalCluster
+
+    from covalent.executor import DaskExecutor
+
+    cluster = LocalCluster()
+
+    mock_app_log = mocker.patch("covalent.executor.executor_plugins.dask.app_log.debug")
+
+    dask_exec = DaskExecutor(cluster.scheduler_address)
+
+    task_metadata = {}
+    job_handle = 42
+
+    result = asyncio.run(dask_exec.cancel(task_metadata, job_handle))
+    mock_app_log.assert_called_with(f"Cancelled future with key {job_handle}")
+    assert result is True
diff --git a/tests/covalent_tests/executor/executor_plugins/local_test.py b/tests/covalent_tests/executor/executor_plugins/local_test.py
index b6a773aa..c30a168a 100644
--- a/tests/covalent_tests/executor/executor_plugins/local_test.py
+++ b/tests/covalent_tests/executor/executor_plugins/local_test.py
@@ -23,11 +23,13 @@
 import io
 import tempfile
 from functools import partial
+from unittest.mock import MagicMock
 
 import pytest
 
 import covalent as ct
 from covalent._shared_files import TaskRuntimeError
+from covalent._shared_files.exceptions import TaskCancelledError
 from covalent._workflow.transport import TransportableObject
 from covalent.executor.base import wrapper_fn
 from covalent.executor.executor_plugins.local import LocalExecutor
@@ -35,7 +37,6 @@ from covalent.executor.executor_plugins.local import LocalExecutor
 
 def test_local_executor_passes_results_dir(mocker):
     """Test that the local executor calls the stream writing function with the results directory specified."""
-
     with tempfile.TemporaryDirectory() as tmp_dir:
 
         @ct.electron
@@ -47,6 +48,11 @@ def test_local_executor_passes_results_dir(mocker):
             "covalent.executor.executor_plugins.local.LocalExecutor.write_streams_to_file"
         )
         le = LocalExecutor()
+        mock_set_job_handle = mocker.patch.object(le, "set_job_handle", MagicMock(return_value=42))
+        mock_get_cancel_requested = mocker.patch.object(
+            le, "get_cancel_requested", MagicMock(return_value=False)
+        )
+        mock__notify = mocker.patch.object(le, "_notify", MagicMock())
 
         assembled_callable = partial(wrapper_fn, TransportableObject(simple_task), [], [])
 
@@ -54,11 +60,14 @@ def test_local_executor_passes_results_dir(mocker):
             function=assembled_callable,
             args=[],
             kwargs={"x": TransportableObject(1), "y": TransportableObject(2)},
-            dispatch_id=-1,
+            dispatch_id="-1",
             results_dir=tmp_dir,
             node_id=0,
         )
         mocked_function.assert_called_once()
+        mock_set_job_handle.assert_called_once()
+        mock_get_cancel_requested.assert_called_once()
+        mock__notify.assert_called_once()
 
 
 def test_local_executor_json_serialization():
@@ -115,12 +124,19 @@ def local_executor_run__mock_task(x):
     return x**2
 
 
-def test_local_executor_run():
+def test_local_executor_run(mocker):
     le = LocalExecutor()
+    mock_set_job_handle = mocker.patch.object(le, "set_job_handle", MagicMock(return_value=42))
+    mock_get_cancel_requested = mocker.patch.object(
+        le, "get_cancel_requested", MagicMock(return_value=False)
+    )
+
     args = [5]
     kwargs = {}
     task_metadata = {"dispatch_id": "asdf", "node_id": 1}
     assert le.run(local_executor_run__mock_task, args, kwargs, task_metadata) == 25
+    mock_set_job_handle.assert_called_once()
+    mock_get_cancel_requested.assert_called_once()
 
 
 def local_executor_run_exception_handling__mock_task(x):
@@ -130,6 +146,10 @@ def local_executor_run_exception_handling__mock_task(x):
 
 def test_local_executor_run_exception_handling(mocker):
     le = LocalExecutor()
+    mock_set_job_handle = mocker.patch.object(le, "set_job_handle", MagicMock(return_value=42))
+    mock_get_cancel_requested = mocker.patch.object(
+        le, "get_cancel_requested", MagicMock(return_value=False)
+    )
     le._task_stdout = io.StringIO()
     le._task_stderr = io.StringIO()
     args = [5]
@@ -139,3 +159,25 @@ def test_local_executor_run_exception_handling(mocker):
         le.run(local_executor_run_exception_handling__mock_task, args, kwargs, task_metadata)
     le._task_stdout.getvalue() == "f output"
     assert "RuntimeError" in le._task_stderr.getvalue()
+
+
+def test_local_executor_get_cancel_requested(mocker):
+    """
+    Test task cancellation request using the local executor
+    """
+    le = LocalExecutor()
+    args = [5]
+    kwargs = {}
+    task_metadata = {"dispatch_id": "asdf", "node_id": 1}
+    le.set_job_handle = MagicMock()
+    le.get_cancel_requested = MagicMock(return_value=True)
+    mock_app_log = mocker.patch("covalent.executor.executor_plugins.local.app_log.debug")
+
+    args = [5]
+    kwargs = {}
+    task_metadata = {"dispatch_id": "asdf", "node_id": 1}
+
+    with pytest.raises(TaskCancelledError):
+        le.run(local_executor_run__mock_task, args, kwargs, task_metadata)
+        le.get_cancel_requested.assert_called_once()
+        assert mock_app_log.call_count == 2
diff --git a/tests/covalent_tests/results_manager_tests/results_manager_test.py b/tests/covalent_tests/results_manager_tests/results_manager_test.py
index e94c5469..33eb7eea 100644
--- a/tests/covalent_tests/results_manager_tests/results_manager_test.py
+++ b/tests/covalent_tests/results_manager_tests/results_manager_test.py
@@ -21,12 +21,12 @@
 """Tests for results manager."""
 
 from http.client import HTTPMessage
-from unittest.mock import ANY, Mock, call
+from unittest.mock import ANY, MagicMock, Mock, call
 
 import pytest
 
 from covalent._results_manager import wait
-from covalent._results_manager.results_manager import _get_result_from_dispatcher
+from covalent._results_manager.results_manager import _get_result_from_dispatcher, cancel
 from covalent._shared_files.config import get_config
 
 DISPATCH_ID = "91c3ee18-5f2d-44ee-ac2a-39b79cf56646"
@@ -65,3 +65,31 @@ def test_get_result_from_dispatcher(mocker, dispatcher_addr):
         ]
         * retries
     )
+
+
+def test_cancel_with_single_task_id(mocker):
+    mock_get_config = mocker.patch("covalent._results_manager.results_manager.get_config")
+    mock_request_post = mocker.patch(
+        "covalent._results_manager.results_manager.requests.post", MagicMock()
+    )
+
+    cancel(dispatch_id="dispatch", task_ids=1)
+
+    assert mock_get_config.call_count == 2
+    mock_request_post.assert_called_once()
+    mock_request_post.return_value.raise_for_status.assert_called_once()
+
+
+def test_cancel_with_multiple_task_ids(mocker):
+    mock_get_config = mocker.patch("covalent._results_manager.results_manager.get_config")
+    mock_task_ids = [0, 1]
+
+    mock_request_post = mocker.patch(
+        "covalent._results_manager.results_manager.requests.post", MagicMock()
+    )
+
+    cancel(dispatch_id="dispatch", task_ids=[1, 2, 3])
+
+    assert mock_get_config.call_count == 2
+    mock_request_post.assert_called_once()
+    mock_request_post.return_value.raise_for_status.assert_called_once()
diff --git a/tests/covalent_tests/workflow/transport_test.py b/tests/covalent_tests/workflow/transport_test.py
index 89af4681..c52ddd0f 100644
--- a/tests/covalent_tests/workflow/transport_test.py
+++ b/tests/covalent_tests/workflow/transport_test.py
@@ -248,8 +248,8 @@ def test_transport_graph_transport_graph_reset(workflow_transport_graph):
     assert list(wtg._graph.nodes) == [0, 1]
     assert list(wtg._graph.edges) == [(0, 1, 0)]
     wtg.reset()
-    assert list(wtg._graph.nodes) == []
-    assert list(wtg._graph.edges) == []
+    assert not list(wtg._graph.nodes)
+    assert not list(wtg._graph.edges)
 
 
 def test_transport_graph_get_and_set_node_values(workflow_transport_graph):
@@ -270,13 +270,13 @@ def test_transport_graph_get_dependencies(workflow_transport_graph):
     """Test the graph node retrieval method in the transport graph."""
 
     wtg = workflow_transport_graph
-    assert list(wtg.get_dependencies(node_key=0)) == []
-    assert list(wtg.get_dependencies(node_key=1)) == []
+    assert not list(wtg.get_dependencies(node_key=0))
+    assert not list(wtg.get_dependencies(node_key=1))
 
     # Add edge relation
     wtg.add_edge(x=0, y=1, edge_name="apples")
 
-    assert list(wtg.get_dependencies(node_key=0)) == []
+    assert not list(wtg.get_dependencies(node_key=0))
     assert list(wtg.get_dependencies(node_key=1)) == [0]
 
 
diff --git a/tests/functional_tests/workflow_cancellation_test.py b/tests/functional_tests/workflow_cancellation_test.py
new file mode 100644
index 00000000..9cbaa6ac
--- /dev/null
+++ b/tests/functional_tests/workflow_cancellation_test.py
@@ -0,0 +1,126 @@
+# Copyright 2021 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+"""End-to-end tests of dispatch cancellation"""
+
+
+import covalent as ct
+import covalent._results_manager.results_manager as rm
+
+TEST_RESULTS_DIR = "/tmp/results"
+
+
+def test_cancel():
+    """Test cancelling dispatch"""
+    import time
+
+    @ct.electron
+    def sleeping_task(delay):
+        import time
+
+        time.sleep(delay)
+        print("Slept for {delay} seconds")
+        return delay
+
+    @ct.lattice
+    def workflow(x):
+        res = sleeping_task(x)
+        res2 = sleeping_task(res)
+        res3 = sleeping_task(res2)
+        return 1
+
+    dispatch_id = ct.dispatch(workflow)(3)
+    time.sleep(0.2)
+
+    ct.cancel(dispatch_id)
+
+    result = ct.get_result(dispatch_id, wait=True)
+    assert result.status == ct.status.CANCELLED
+    rm._delete_result(dispatch_id)
+
+
+def test_cancel_task():
+    """Test cancelling specific task"""
+    import time
+
+    @ct.electron
+    def sleeping_task(delay):
+        import time
+
+        time.sleep(delay)
+        print("Slept for {delay} seconds")
+        return delay
+
+    @ct.lattice
+    def workflow(x):
+        res = sleeping_task(x)
+        res2 = sleeping_task(res)
+
+        return 1
+
+    dispatch_id = ct.dispatch(workflow)(3)
+    time.sleep(0.1)
+
+    ct.cancel(dispatch_id, task_ids=[2])
+
+    result = ct.get_result(dispatch_id, wait=True)
+    assert result.get_node_result(0)["status"] == ct.status.COMPLETED
+    assert result.get_node_result(2)["status"] == ct.status.CANCELLED
+    assert result.status == ct.status.CANCELLED
+    rm._delete_result(dispatch_id)
+
+
+def test_cancel_sublattice():
+    """Test cancelling sublattice"""
+    import time
+
+    @ct.electron
+    def sleeping_task(delay):
+        import time
+
+        time.sleep(delay)
+        print("Slept for {delay} seconds")
+        return delay
+
+    @ct.electron
+    @ct.lattice
+    def sub_workflow(x):
+        res = sleeping_task(x)
+        res2 = sleeping_task(res)
+
+        return 1
+
+    @ct.lattice
+    def workflow(x):
+        return sub_workflow(3)
+
+    dispatch_id = ct.dispatch(workflow)(3)
+    time.sleep(0.5)
+
+    ct.cancel(dispatch_id, task_ids=[0])
+
+    result = ct.get_result(dispatch_id, wait=True)
+
+    tg = result.lattice.transport_graph
+    sub_dispatch_id = tg.get_node_value(0, "sub_dispatch_id")
+
+    print("Sublattice dispatch id:", sub_dispatch_id)
+    sub_res = ct.get_result(sub_dispatch_id)
+    assert sub_res.status == ct.status.CANCELLED