Repository: DataBiosphere/toil

Issue:
Keep workflow metadata between Toil server container restarts, on AGC
AGC gets mad if an engine ever forgets about a workflow that ran.

Toil _can_ remember workflows between server invocations, if its workflow directories are kept around. But right now the Toil server container we use for AGC keeps all its state in the container, and it is all lost of the container ever restarts.

We need to set up the Toil AGC engine container to keep those workflow directories on something persistent, like Amazon EFS (which is NFS AFAIK). If those directories are also used as scratch during the workflow run, either we need to make sure that scratch work is safe to do on EFS, or we need to move it away from the metadata.

If the metadata and output files are not happy being stored on EFS, then we need to replace them with something (sqlite with appropriate precautions?) that is happy being backed by EFS.

┆Issue is synchronized with this [Jira Story](https://ucsc-cgl.atlassian.net/browse/TOIL-1148)
┆epic: AGC
┆friendlyId: TOIL-1148


Discussion:
adamnovak: I've revised the WES server tests substantially; they can now fake Celery with `multiprocessing` instead, so we can actually run a workflow on the server in the tests without having a Celery broker/container handy.
adamnovak: @w-gao Since Lon is out this week, can you review this?
w-gao: Looks like our docker build is failing because rpm.aventer.biz changed their directory structure today...?

Maybe it'll work if we change `https://rpm.aventer.biz/Ubuntu/dists/focal/binary-amd64/mesos-1.11.x.deb` -> `https://rpm.aventer.biz/Ubuntu/dists/focal/main/binary-amd64/Packages/mesos-1.11.x.deb` in our Dockerfile.
adamnovak: The Docker build failed last time when it was trying to download packages; I'm not sure why. I couldn't replicate it. I touched the Dockerfile so maybe it will work this time?
adamnovak: Looks like we got another:

```
#10 220.1 E: Failed to fetch http://ports.ubuntu.com/ubuntu-ports/pool/main/liba/libalgorithm-merge-perl/libalgorithm-merge-perl_0.08-3_all.deb  Undetermined Error [IP: 185.125.190.39 80]
```
adamnovak: That URL exists for me; maybe something is going wrong with networking from our Docker builds? I've restarted that appliance build, so hopefully it works this time?

Patch:
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index e5aaf392..23a3fd0d 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -36,7 +36,7 @@ before_script:
   - sudo apt-get install -y software-properties-common build-essential virtualenv
   - sudo add-apt-repository -y ppa:deadsnakes/ppa
   - sudo apt-get update
-  - sudo apt-get install -y tzdata jq python3.7 python3.7-dev python3.8 python3.8-dev python3.9 python3.9-dev python3.9-distutils  # python3.10 python3.10-dev python3.10-venv
+  - sudo apt-get install -y tzdata jq python3.7 python3.7-dev python3.7-venv python3.8 python3.8-dev python3.8-venv python3.9 python3.9-dev python3.9-venv python3.9-distutils  # python3.10 python3.10-dev python3.10-venv
 
 after_script:
   # We need to clean up any files that Toil may have made via Docker that
@@ -58,7 +58,7 @@ lint:
   stage: linting_and_dependencies
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && make prepare && make develop extras=[all] packages=htcondor
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && make prepare && make develop extras=[all] packages=htcondor
     - make mypy
     - make docs
     # - make diff_pydocstyle_report
@@ -68,7 +68,7 @@ cwl_dependency_is_stand_alone:
   stage: linting_and_dependencies
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && make prepare && make develop extras=[cwl]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && make prepare && make develop extras=[cwl]
     - make test tests=src/toil/test/docs/scriptsTest.py::ToilDocumentationTest::testCwlexample
 
 
@@ -76,13 +76,13 @@ wdl_dependency_is_stand_alone:
   stage: linting_and_dependencies
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && make prepare && make develop extras=[wdl]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && make prepare && make develop extras=[wdl]
     - make test tests=src/toil/test/wdl/toilwdlTest.py::ToilWdlIntegrationTest::testMD5sum
 
 quick_test_offline:
   stage: basic_tests
   script:
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv
     - . venv/bin/activate
     - pip install -U pip wheel
     - make prepare
@@ -94,7 +94,7 @@ py37_main:
   stage: basic_tests
   script:
     - pwd
-    - virtualenv -p python3.7 venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
+    - python3.7 -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
     - make test tests=src/toil/test/src
     - make test tests=src/toil/test/utils
     - TOIL_SKIP_DOCKER=true make test tests=src/toil/test/lib
@@ -103,7 +103,7 @@ py37_appliance_build:
   stage: basic_tests
   script:
     - pwd
-    - virtualenv -p python3.7 venv && . venv/bin/activate && pip install -U pip wheel && make prepare && pip install pycparser && make develop extras=[all] packages='htcondor awscli'
+    - python3.7 -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && pip install pycparser && make develop extras=[all] packages='htcondor awscli'
     # This reads GITLAB_SECRET_FILE_QUAY_CREDENTIALS
     - python setup_gitlab_docker.py
     - make push_docker
@@ -113,7 +113,7 @@ py38_main:
   stage: basic_tests
   script:
     - pwd
-    - virtualenv -p python3.8 venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
+    - python3.8 -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
     - make test tests=src/toil/test/src
     - make test tests=src/toil/test/utils
     - TOIL_SKIP_DOCKER=true make test tests=src/toil/test/lib
@@ -122,7 +122,7 @@ py38_appliance_build:
   stage: basic_tests
   script:
     - pwd
-    - virtualenv -p python3.8 venv && . venv/bin/activate && pip install -U pip wheel && make prepare && pip install pycparser && make develop extras=[all] packages='htcondor awscli'
+    - python3.8 -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && pip install pycparser && make develop extras=[all] packages='htcondor awscli'
     # This reads GITLAB_SECRET_FILE_QUAY_CREDENTIALS
     - python setup_gitlab_docker.py
     - make push_docker
@@ -132,7 +132,7 @@ py39_main:
   stage: basic_tests
   script:
     - pwd
-    - virtualenv -p python3.9 venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages=htcondor
+    - python3.9 -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages=htcondor
     - make test tests=src/toil/test/src
     - make test tests=src/toil/test/utils
     - TOIL_SKIP_DOCKER=true make test tests=src/toil/test/lib
@@ -141,7 +141,7 @@ py39_appliance_build:
   stage: basic_tests
   script:
     - pwd
-    - virtualenv -p python3.9 venv && . venv/bin/activate && pip install -U pip wheel && make prepare && pip install pycparser && make develop extras=[all] packages='htcondor awscli'
+    - python3.9 -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && pip install pycparser && make develop extras=[all] packages='htcondor awscli'
     # This reads GITLAB_SECRET_FILE_QUAY_CREDENTIALS
     - python setup_gitlab_docker.py
     - make push_docker
@@ -150,7 +150,7 @@ batch_systems:
   stage: main_tests
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
     - wget https://github.com/ohsu-comp-bio/funnel/releases/download/0.10.1/funnel-linux-amd64-0.10.1.tar.gz
     - tar -xvf funnel-linux-amd64-0.10.1.tar.gz funnel
     - export FUNNEL_SERVER_USER=toil
@@ -184,7 +184,7 @@ cwl_v1.0:
   only: []
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws]
     - mypy --ignore-missing-imports --no-strict-optional $(pwd)/src/toil/cwl/cwltoil.py  # make this a separate linting stage
     - python setup_gitlab_docker.py  # login to increase the docker.io rate limit
     - make test tests=src/toil/test/cwl/cwlTest.py::CWLv10Test
@@ -194,7 +194,7 @@ cwl_v1.1:
   only: []
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws]
     - python setup_gitlab_docker.py  # login to increase the docker.io rate limit
     - make test tests=src/toil/test/cwl/cwlTest.py::CWLv11Test
 
@@ -202,7 +202,7 @@ cwl_v1.2:
   stage: main_tests
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws]
     - python setup_gitlab_docker.py  # login to increase the docker.io rate limit
     - make test tests=src/toil/test/cwl/cwlTest.py::CWLv12Test
 
@@ -210,7 +210,7 @@ cwl_on_arm:
   stage: main_tests
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws]
     - python setup_gitlab_docker.py  # login to increase the docker.io rate limit
     # This reads GITLAB_SECRET_FILE_SSH_KEYS
     - python setup_gitlab_ssh.py
@@ -222,7 +222,7 @@ cwl_v1.0_kubernetes:
   only: []
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws,kubernetes]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws,kubernetes]
     - export TOIL_KUBERNETES_OWNER=toiltest
     - export TOIL_AWS_SECRET_NAME=shared-s3-credentials
     - export TOIL_KUBERNETES_HOST_PATH=/data/scratch
@@ -238,7 +238,7 @@ cwl_v1.1_kubernetes:
   only: []
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws,kubernetes]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws,kubernetes]
     - export TOIL_KUBERNETES_OWNER=toiltest
     - export TOIL_AWS_SECRET_NAME=shared-s3-credentials
     - export TOIL_KUBERNETES_HOST_PATH=/data/scratch
@@ -252,7 +252,7 @@ cwl_v1.2_kubernetes:
   stage: main_tests
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws,kubernetes]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[cwl,aws,kubernetes]
     - export TOIL_KUBERNETES_OWNER=toiltest
     - export TOIL_AWS_SECRET_NAME=shared-s3-credentials
     - export TOIL_KUBERNETES_HOST_PATH=/data/scratch
@@ -274,7 +274,7 @@ wdl:
   script:
     - pwd
     - apt update && apt install -y default-jre
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all]
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all]
     - which java &> /dev/null || { echo >&2 "Java is not installed.  Install java to run these tests."; exit 1; }
     - make test tests=src/toil/test/wdl/toilwdlTest.py  # needs java (default-jre) to run "GATK.jar"
     - make test tests=src/toil/test/wdl/builtinTest.py
@@ -283,7 +283,7 @@ jobstore_and_provisioning:
   stage: main_tests
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages=htcondor
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages=htcondor
     - make test tests=src/toil/test/jobStores/jobStoreTest.py
     - make test tests=src/toil/test/sort/sortTest.py
     - make test tests=src/toil/test/provisioners/aws/awsProvisionerTest.py
@@ -296,20 +296,23 @@ integration:
   stage: integration
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
     - export TOIL_TEST_INTEGRATIVE=True
     - export TOIL_AWS_KEYNAME=id_rsa
     - export TOIL_AWS_ZONE=us-west-2a
     # This reads GITLAB_SECRET_FILE_SSH_KEYS
     - python setup_gitlab_ssh.py
     - chmod 400 /root/.ssh/id_rsa
+    # Test integration with job stores
     - make test tests=src/toil/test/jobStores/jobStoreTest.py
+    # Test server and its integration with AWS
+    - make test tests=src/toil/test/server
 
 provisioner_integration:
   stage: integration
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
     - python setup_gitlab_ssh.py && chmod 400 /root/.ssh/id_rsa
     - echo $'Host *\n    AddressFamily inet' > /root/.ssh/config
     - export LIBPROCESS_IP=127.0.0.1
@@ -327,7 +330,7 @@ google_jobstore:
   stage: integration
   script:
     - pwd
-    - virtualenv -p ${MAIN_PYTHON_PKG} venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
+    - ${MAIN_PYTHON_PKG} -m virtualenv venv && . venv/bin/activate && pip install -U pip wheel && make prepare && make develop extras=[all] packages='htcondor awscli'
     - python setup_gitlab_ssh.py && chmod 400 /root/.ssh/id_rsa
     - echo $'Host *\n    AddressFamily inet' > /root/.ssh/config
     - export LIBPROCESS_IP=127.0.0.1
@@ -342,7 +345,7 @@ cactus_integration:
   stage: integration
   script:
     - set -e
-    - virtualenv --system-site-packages --python ${MAIN_PYTHON_PKG} venv
+    - ${MAIN_PYTHON_PKG} -m virtualenv --system-site-packages venv
     - . venv/bin/activate
     - pip install -U pip wheel
     - pip install .[aws,kubernetes]
diff --git a/contrib/admin/cleanup_aws_resources.py b/contrib/admin/cleanup_aws_resources.py
index 7310c5ef..48e9aeb3 100755
--- a/contrib/admin/cleanup_aws_resources.py
+++ b/contrib/admin/cleanup_aws_resources.py
@@ -21,6 +21,7 @@ pkg_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
 sys.path.insert(0, pkg_root)  # noqa
 
 from src.toil.lib import aws
+from src.toil.lib.aws import session
 from src.toil.lib.aws.utils import delete_iam_role, delete_iam_instance_profile, delete_s3_bucket, delete_sdb_domain
 from src.toil.lib.generatedEC2Lists import regionDict
 
@@ -278,7 +279,8 @@ def main(argv):
                 if response.lower() in ('y', 'yes'):
                     print('\nOkay, now deleting...')
                     for bucket, region in buckets.items():
-                        delete_s3_bucket(bucket, region)
+                        s3_resource = session.resource('s3', region_name=region)
+                        delete_s3_bucket(s3_resource, bucket)
                     print('S3 Bucket Deletions Successful.')
 
     if not options.skip_sdb:
diff --git a/docker/Dockerfile.py b/docker/Dockerfile.py
index 454ab9b5..52e6e74f 100644
--- a/docker/Dockerfile.py
+++ b/docker/Dockerfile.py
@@ -28,6 +28,7 @@ pip = f'{python} -m pip'
 dependencies = ' '.join(['libffi-dev',  # For client side encryption for extras with PyNACL
                          python,
                          f'{python}-dev',
+                         'python3.7-distutils' if python == 'python3.7' else '',
                          'python3.8-distutils' if python == 'python3.8' else '',
                          'python3.9-distutils' if python == 'python3.9' else '',
                          # 'python3.9-venv' if python == 'python3.9' else '',
@@ -85,32 +86,42 @@ print(heredoc('''
 
     ARG TARGETARCH
 
+    RUN if [ -z "$TARGETARCH" ] ; then echo "Specify a TARGETARCH argument to build this container"; exit 1; fi
+
     # make sure we don't use too new a version of setuptools (which can get out of sync with poetry and break things)
     ENV SETUPTOOLS_USE_DISTUTILS=stdlib
 
-    RUN apt-get -y update --fix-missing && apt-get -y upgrade && apt-get -y install apt-transport-https ca-certificates software-properties-common && apt-get clean && rm -rf /var/lib/apt/lists/*
+    # Try to avoid "Failed to fetch ...  Undetermined Error" from apt
+    # See <https://stackoverflow.com/a/66523384>
+    RUN printf 'Acquire::http::Pipeline-Depth "0";\\nAcquire::http::No-Cache=True;\\nAcquire::BrokenProxy=true;\\n' >/etc/apt/apt.conf.d/99fixbadproxy
+
+    RUN apt-get -y update --fix-missing && apt-get -y upgrade && apt-get -y install apt-transport-https ca-certificates software-properties-common curl && apt-get clean && rm -rf /var/lib/apt/lists/*
 
     RUN add-apt-repository -y ppa:deadsnakes/ppa
 
+    # Find a repo with a Mesos build.
+    # See https://rpm.aventer.biz/README.txt
+    # A working snapshot is https://ipfs.io/ipfs/QmfTy9sXhHsgyWwosCJDfYR4fChTosA8HhoaMgmeJ5LSmS/
+    # As archived with:
+    # mkdir mesos-repo && cd mesos-repo
+    # wget --recursive --restrict-file-names=windows -k --convert-links --no-parent --page-requisites https://rpm.aventer.biz/Ubuntu/ https://www.aventer.biz/assets/support_aventer.asc https://rpm.aventer.biz/README.txt
+    # ipfs add -r .
+    RUN echo "deb https://rpm.aventer.biz/Ubuntu focal main" \
+        > /etc/apt/sources.list.d/mesos.list \
+        && curl https://www.aventer.biz/assets/support_aventer.asc | apt-key add -
+
     RUN apt-get -y update --fix-missing && \
         DEBIAN_FRONTEND=noninteractive apt-get -y upgrade && \
         DEBIAN_FRONTEND=noninteractive apt-get -y install {dependencies} && \
+        if [ $TARGETARCH = amd64 ] ; then DEBIAN_FRONTEND=noninteractive apt-get -y install mesos ; mesos-agent --help >/dev/null ; fi && \
         apt-get clean && \
         rm -rf /var/lib/apt/lists/*
     
-    # Install a Mesos build from somewhere and test it.
-    # This is /ipfs/QmRCNmVVrWPPQiEw2PrFLmb8ps6oETQvtKv8dLVN8ZRwFz/mesos-1.11.x.deb
-    RUN if [ $TARGETARCH = amd64 ] ; then \
-        wget -q https://rpm.aventer.biz/Ubuntu/dists/focal/binary-amd64/mesos-1.11.x.deb && \
-        dpkg -i mesos-1.11.x.deb && \
-        rm mesos-1.11.x.deb && \
-        mesos-agent --help >/dev/null ; \
-        fi
-    
     # Install a particular old Debian Sid Singularity from somewhere.
     ADD singularity-sources.tsv /etc/singularity/singularity-sources.tsv
     RUN wget -q "$(cat /etc/singularity/singularity-sources.tsv | grep "^$TARGETARCH" | cut -f3)" && \
         dpkg -i singularity-container_3*.deb && \
+        rm singularity-container_3*.deb && \
         sed -i 's!bind path = /etc/localtime!#bind path = /etc/localtime!g' /etc/singularity/singularity.conf && \
         mkdir -p /usr/local/libexec/toil && \
         mv /usr/bin/singularity /usr/local/libexec/toil/singularity-real \
diff --git a/docs/running/server/wes.rst b/docs/running/server/wes.rst
index dd8f4add..434ac3e0 100644
--- a/docs/running/server/wes.rst
+++ b/docs/running/server/wes.rst
@@ -78,6 +78,8 @@ Below is a detailed summary of all available options:
 --dest_bucket_base DEST_BUCKET_BASE
             Direct CWL workflows to save output files to dynamically generated unique paths under the given URL.
             Supports AWS S3.
+--state_store STATE_STORE
+            The local path or S3 URL where workflow state metadata should be stored. (default: in --work_dir)
 
 .. _GA4GH docs on CORS: https://w3id.org/ga4gh/product-approval-support/cors
 
diff --git a/src/toil/common.py b/src/toil/common.py
index 23f28803..dc84d31b 100644
--- a/src/toil/common.py
+++ b/src/toil/common.py
@@ -28,6 +28,7 @@ from argparse import (
     Namespace,
     _ArgumentGroup,
 )
+from functools import lru_cache
 from types import TracebackType
 from typing import (
     IO,
@@ -765,7 +766,7 @@ def parseBool(val: str) -> bool:
     else:
         raise RuntimeError("Could not interpret \"%s\" as a boolean value" % val)
 
-
+@lru_cache(maxsize=None)
 def getNodeID() -> str:
     """
     Return unique ID of the current node (host). The resulting string will be convertable to a uuid.UUID.
@@ -1227,7 +1228,7 @@ class Toil(ContextManager["Toil"]):
         if not os.path.exists(workDir):
             raise RuntimeError(f'The directory specified by --workDir or TOIL_WORKDIR ({workDir}) does not exist.')
         return workDir
-        
+
     @classmethod
     def get_toil_coordination_dir(cls, configWorkDir: Optional[str] = None) -> str:
         """
@@ -1237,7 +1238,7 @@ class Toil(ContextManager["Toil"]):
         :param configWorkDir: Value passed to the program using the --workDir flag
         :return: Path to the Toil coordination directory.
         """
-        
+
         # Get our user ID
         user_id = os.getuid()
         in_memory_base = os.path.join('/var/run/user', str(user_id), 'toil')
@@ -1251,7 +1252,7 @@ class Toil(ContextManager["Toil"]):
                 return in_memory_base
             except:
                 pass
-        
+
         # Otherwise use the on-disk one.
         return cls.getToilWorkDir(configWorkDir)
 
@@ -1260,11 +1261,11 @@ class Toil(ContextManager["Toil"]):
     def _get_workflow_path_component(workflow_id: str) -> str:
         """
         Get a safe filesystem path component for a workflow.
-        
+
         Will be consistent for all processes on a given machine, and different
         for all processes on different machines.
-        
-        :param workflow_id: THe ID of the current Toil workflow.
+
+        :param workflow_id: The ID of the current Toil workflow.
         """
         return str(uuid.uuid5(uuid.UUID(getNodeID()), workflow_id)).replace('-', '')
 
@@ -1294,7 +1295,7 @@ class Toil(ContextManager["Toil"]):
         else:
             logger.debug('Created the workflow directory for this machine at %s' % workflowDir)
         return workflowDir
-        
+
     @classmethod
     def get_local_workflow_coordination_dir(
         cls, workflow_id: str, config_work_dir: Optional[str] = None
@@ -1303,22 +1304,22 @@ class Toil(ContextManager["Toil"]):
         Return the directory where coordination files should be located for
         this workflow on this machine. These include internal Toil databases
         and lock files for the machine.
-        
+
         If an in-memory filesystem is available, it is used. Otherwise, the
         local workflow directory, which may be on a shared network filesystem,
         is used.
-        
+
         :param workflow_id: Unique ID of the current workflow.
         :param config_work_dir: Value used for the work directory in the
                current Toil Config.
-        
+
         :return: Path to the local workflow coordination directory on this
                  machine.
         """
-        
+
         # Start with the base coordination or work dir
         base = cls.get_toil_coordination_dir(config_work_dir)
-        
+
         # Make a per-workflow and node subdirectory
         subdir = os.path.join(base, cls._get_workflow_path_component(workflow_id))
         # Make it exist
@@ -1326,8 +1327,8 @@ class Toil(ContextManager["Toil"]):
         # TODO: May interfere with workflow directory creation logging if it's the same directory.
         # Return it
         return subdir
-            
-                
+
+
 
     def _runMainLoop(self, rootJob: "JobDescription") -> Any:
         """
diff --git a/src/toil/lib/aws/utils.py b/src/toil/lib/aws/utils.py
index c0568b5b..0e79a540 100644
--- a/src/toil/lib/aws/utils.py
+++ b/src/toil/lib/aws/utils.py
@@ -153,15 +153,17 @@ def retry_s3(delays: Iterable[float] = DEFAULT_DELAYS, timeout: float = DEFAULT_
     return old_retry(delays=delays, timeout=timeout, predicate=predicate)
 
 @retry(errors=[BotoServerError])
-def delete_s3_bucket(bucket: str, region: Optional[str], quiet: bool = True) -> None:
+def delete_s3_bucket(
+    s3_resource: "S3ServiceResource",
+    bucket: str,
+    quiet: bool = True
+) -> None:
     """
     Delete the given S3 bucket.
     """
-    printq(f'Deleting s3 bucket in region "{region}": {bucket}', quiet)
-    s3_client = cast(S3Client, session.client('s3', region_name=region))
-    s3_resource = cast(S3ServiceResource, session.resource('s3', region_name=region))
+    printq(f'Deleting s3 bucket: {bucket}', quiet)
 
-    paginator = s3_client.get_paginator('list_object_versions')
+    paginator = s3_resource.meta.client.get_paginator('list_object_versions')
     try:
         for response in paginator.paginate(Bucket=bucket):
             # Versions and delete markers can both go in here to be deleted.
@@ -173,15 +175,15 @@ def delete_s3_bucket(bucket: str, region: Optional[str], quiet: bool = True) ->
                                               cast(List[Dict[str, Any]], response.get('DeleteMarkers', []))
             for entry in to_delete:
                 printq(f"    Deleting {entry['Key']} version {entry['VersionId']}", quiet)
-                s3_client.delete_object(Bucket=bucket, Key=entry['Key'], VersionId=entry['VersionId'])
+                s3_resource.meta.client.delete_object(Bucket=bucket, Key=entry['Key'], VersionId=entry['VersionId'])
         s3_resource.Bucket(bucket).delete()
         printq(f'\n * Deleted s3 bucket successfully: {bucket}\n\n', quiet)
-    except s3_client.exceptions.NoSuchBucket:
+    except s3_resource.meta.client.exceptions.NoSuchBucket:
         printq(f'\n * S3 bucket no longer exists: {bucket}\n\n', quiet)
 
 
 def create_s3_bucket(
-    s3_session: "S3ServiceResource",
+    s3_resource: "S3ServiceResource",
     bucket_name: str,
     region: Union["BucketLocationConstraintType", Literal["us-east-1"]],
 ) -> "Bucket":
@@ -195,9 +197,9 @@ def create_s3_bucket(
     """
     logger.debug("Creating bucket '%s' in region %s.", bucket_name, region)
     if region == "us-east-1":  # see https://github.com/boto/boto3/issues/125
-        bucket = s3_session.create_bucket(Bucket=bucket_name)
+        bucket = s3_resource.create_bucket(Bucket=bucket_name)
     else:
-        bucket = s3_session.create_bucket(
+        bucket = s3_resource.create_bucket(
             Bucket=bucket_name,
             CreateBucketConfiguration={"LocationConstraint": region},
         )
diff --git a/src/toil/provisioners/abstractProvisioner.py b/src/toil/provisioners/abstractProvisioner.py
index c039c8b9..7712cb31 100644
--- a/src/toil/provisioners/abstractProvisioner.py
+++ b/src/toil/provisioners/abstractProvisioner.py
@@ -731,10 +731,10 @@ class AbstractProvisioner(ABC):
         # mesos-agent. If there are multiple keys to be transferred, then the last one to be transferred must be
         # set to keyPath.
         MESOS_LOG_DIR = '--log_dir=/var/lib/mesos '
-        LEADER_DOCKER_ARGS = '--webui_dir=/share/mesos/webui --registry=in_memory --cluster={name}'
+        LEADER_DOCKER_ARGS = '--registry=in_memory --cluster={name}'
         # --no-systemd_enable_support is necessary in Ubuntu 16.04 (otherwise,
         # Mesos attempts to contact systemd but can't find its run file)
-        WORKER_DOCKER_ARGS = '--launcher_dir=/libexec/mesos --work_dir=/var/lib/mesos --master={ip}:5050 --attributes=preemptable:{preemptable} --no-hostname_lookup --no-systemd_enable_support'
+        WORKER_DOCKER_ARGS = '--work_dir=/var/lib/mesos --master={ip}:5050 --attributes=preemptable:{preemptable} --no-hostname_lookup --no-systemd_enable_support'
 
         if self.clusterType == 'mesos':
             if role == 'leader':
diff --git a/src/toil/server/app.py b/src/toil/server/app.py
index 47fe5306..4a12638d 100644
--- a/src/toil/server/app.py
+++ b/src/toil/server/app.py
@@ -15,6 +15,8 @@ import argparse
 import logging
 import os
 
+from typing import Type
+
 import connexion  # type: ignore
 
 from toil.lib.misc import get_public_ip
@@ -30,6 +32,9 @@ def parser_with_server_options() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="Toil server mode.")
 
     parser.add_argument("--debug", action="store_true", default=False)
+    parser.add_argument("--bypass_celery", action="store_true", default=False,
+                        help="Skip sending workflows to Celery and just run them under the"
+                             "server. For testing.")
     parser.add_argument("--host", type=str, default="127.0.0.1",
                         help="The host interface that the Toil server binds on. (default: '127.0.0.1').")
     parser.add_argument("--port", type=int, default=8080,
@@ -53,7 +58,10 @@ def parser_with_server_options() -> argparse.ArgumentParser:
     parser.add_argument("--work_dir", type=str, default=os.path.join(os.getcwd(), "workflows"),
                         help="The directory where workflows should be stored. This directory should be "
                              "empty or only contain previous workflows. (default: './workflows').")
-    parser.add_argument("--opt", "-o", type=str, action="append",
+    parser.add_argument("--state_store", type=str, default=None,
+                        help="The local path or S3 URL where workflow state metadata should be stored. "
+                             "(default: in --work_dir)")
+    parser.add_argument("--opt", "-o", type=str, action="append", default=[],
                         help="Specify the default parameters to be sent to the workflow engine for each "
                              "run. Options taking arguments must use = syntax. Accepts multiple values.\n"
                              "Example: '--opt=--logLevel=CRITICAL --opt=--workDir=/tmp'.")
@@ -80,7 +88,11 @@ def create_app(args: argparse.Namespace) -> "connexion.FlaskApp":
         CORS(flask_app.app, resources={r"/ga4gh/*": {"origins": args.cors_origins}})
 
     # add workflow execution service (WES) API endpoints
-    backend = ToilBackend(work_dir=args.work_dir, options=args.opt, dest_bucket_base=args.dest_bucket_base)
+    backend = ToilBackend(work_dir=args.work_dir,
+                          state_store=args.state_store,
+                          options=args.opt,
+                          dest_bucket_base=args.dest_bucket_base,
+                          bypass_celery=args.bypass_celery)
 
     flask_app.add_api('workflow_execution_service.swagger.yaml',
                       resolver=connexion.Resolver(backend.resolve_operation_id))  # noqa
@@ -95,13 +107,13 @@ def create_app(args: argparse.Namespace) -> "connexion.FlaskApp":
         flask_app.app.add_url_rule("/engine/v1/status", view_func=backend.get_health)
         # And we can provide lost humans some information on what they are looking at
         flask_app.app.add_url_rule("/", view_func=backend.get_homepage)
-
+        
     return flask_app
 
 
 def start_server(args: argparse.Namespace) -> None:
     """ Start a Toil server."""
-    
+
     # Explain a bit about who and where we are
     logger.info("Toil WES server version %s starting...", version)
     if running_on_ecs():
@@ -116,7 +128,7 @@ def start_server(args: argparse.Namespace) -> None:
 
     host = args.host
     port = args.port
-    
+
     if args.debug:
         flask_app.run(host=host, port=port)
     else:
diff --git a/src/toil/server/utils.py b/src/toil/server/utils.py
index 6a381890..00c47dce 100644
--- a/src/toil/server/utils.py
+++ b/src/toil/server/utils.py
@@ -13,20 +13,32 @@
 # limitations under the License.
 import fcntl
 import os
+from abc import abstractmethod
 from datetime import datetime
-from typing import Optional
+from typing import Dict, Optional, Tuple
 from urllib.parse import urlparse
 
+import logging
 import requests
 
 from toil.lib.retry import retry
+from toil.lib.io import AtomicFileCreate
 
+try:
+    from toil.lib.aws import get_current_aws_region
+    from toil.lib.aws.session import client
+    from toil.lib.aws.utils import retry_s3
+    HAVE_S3 = True
+except ImportError:
+    HAVE_S3 = False
+
+logger = logging.getLogger(__name__)
 
 def get_iso_time() -> str:
     """
     Return the current time in ISO 8601 format.
     """
-    return datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
+    return datetime.now().isoformat()
 
 
 def link_file(src: str, dest: str) -> None:
@@ -85,7 +97,6 @@ def get_file_class(path: str) -> str:
         return "Directory"
     return "Unknown"
 
-
 @retry(errors=[OSError, BlockingIOError])
 def safe_read_file(file: str) -> Optional[str]:
     """
@@ -115,18 +126,504 @@ def safe_write_file(file: str, s: str) -> None:
     Safely write to a file by acquiring an exclusive lock to prevent other
     processes from reading and writing to it while writing.
     """
-    # Open in read and update mode, so we don't modify the file before we acquire a lock
-    file_obj = open(file, "r+")
 
-    try:
-        # acquire an exclusive lock
-        fcntl.flock(file_obj.fileno(), fcntl.LOCK_EX)
+    if os.path.exists(file):
+        # Going to overwrite without anyone else being able to see this
+        # intermediate state.
+
+        # Open in read and update mode, so we don't modify the file before we acquire a lock
+        file_obj = open(file, "r+")
 
         try:
-            file_obj.seek(0)
-            file_obj.write(s)
-            file_obj.truncate()
+            # acquire an exclusive lock
+            fcntl.flock(file_obj.fileno(), fcntl.LOCK_EX)
+
+            try:
+                file_obj.seek(0)
+                file_obj.write(s)
+                file_obj.truncate()
+            finally:
+                fcntl.flock(file_obj.fileno(), fcntl.LOCK_UN)
         finally:
-            fcntl.flock(file_obj.fileno(), fcntl.LOCK_UN)
-    finally:
-        file_obj.close()
+            file_obj.close()
+    else:
+        # Contend with everyone else to create the file. Last write will win
+        # but it will be atomic because of the filesystem.
+        with AtomicFileCreate(file) as temp_name:
+            with open(temp_name, "w") as file_obj:
+                file_obj.write(s)
+
+class MemoryStateCache:
+    """
+    An in-memory place to store workflow state.
+    """
+
+    def __init__(self) -> None:
+        """
+        Make a new in-memory state cache.
+        """
+
+        super().__init__()
+        self._data: Dict[Tuple[str, str], Optional[str]] = {}
+
+    def get(self, workflow_id: str, key: str) -> Optional[str]:
+        """
+        Get a key value from memory.
+        """
+        return self._data.get((workflow_id, key))
+
+    def set(self, workflow_id: str, key: str, value: Optional[str]) -> None:
+        """
+        Set or clear a key value in memory.
+        """
+
+        if value is None:
+            try:
+                del self._data[(workflow_id, key)]
+            except KeyError:
+                pass
+        else:
+            self._data[(workflow_id, key)] = value
+
+class AbstractStateStore:
+    """
+    A place for the WES server to keep its state: the set of workflows that
+    exist and whether they are done or not.
+
+    This is a key-value store, with keys namespaced by workflow ID. Concurrent
+    access from multiple threads or processes is safe and globally consistent.
+
+    Keys and workflow IDs are restricted to [-a-zA-Z0-9_], because backends may
+    use them as path or URL components.
+
+    Key values are either a string, or None if the key is not set.
+
+    Workflow existence isn't a thing; nonexistent workflows just have None for
+    all keys.
+
+    Note that we don't yet have a cleanup operation: things are stored
+    permanently. Even clearing all the keys may leave data behind.
+
+    Also handles storage for a local cache, with a separate key namespace (not
+    a read/write-through cache).
+
+    TODO: Can we replace this with just using a JobStore eventually, when
+    AWSJobStore no longer needs SimpleDB?
+    """
+
+    def __init__(self):
+        """
+        Set up the AbstractStateStore and its cache.
+        """
+
+        # We maintain a local cache here.
+        # TODO: Upgrade to an LRU cache wehn we finally learn to paginate
+        # workflow status
+        self._cache = MemoryStateCache()
+
+    @abstractmethod
+    def get(self, workflow_id: str, key: str) -> Optional[str]:
+        """
+        Get the value of the given key for the given workflow, or None if the
+        key is not set for the workflow.
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def set(self, workflow_id: str, key: str, value: Optional[str]) -> None:
+        """
+        Set the value of the given key for the given workflow. If the value is
+        None, clear the key.
+        """
+        raise NotImplementedError
+
+    def read_cache(self, workflow_id: str, key: str) -> Optional[str]:
+        """
+        Read a value from a local cache, without checking the actual backend.
+        """
+
+        return self._cache.get(workflow_id, key)
+
+    def write_cache(self, workflow_id: str, key: str, value: Optional[str]) -> None:
+        """
+        Write a value to a local cache, without modifying the actual backend.
+        """
+        self._cache.set(workflow_id, key, value)
+
+class MemoryStateStore(MemoryStateCache, AbstractStateStore):
+    """
+    An in-memory place to store workflow state, for testing.
+
+    Inherits from MemoryStateCache first to provide implementations for
+    AbstractStateStore.
+    """
+
+    def __init__(self):
+        super().__init__()
+
+class FileStateStore(AbstractStateStore):
+    """
+    A place to store workflow state that uses a POSIX-compatible file system.
+    """
+
+    def __init__(self, url: str) -> None:
+        """
+        Connect to the state store in the given local directory.
+
+        :param url: Local state store path. Interpreted as a URL, so can't
+                    contain ? or #.
+        """
+        super().__init__()
+        parse = urlparse(url)
+        if parse.scheme.lower() not in ['file', '']:
+            # We want to catch if we get the wrong argument.
+            raise RuntimeError(f"{url} doesn't look like a local path")
+        if not os.path.exists(parse.path):
+            # We need this directory to exist.
+            os.makedirs(parse.path, exist_ok=True)
+        logger.debug("Connected to FileStateStore at %s", url)
+        self._base_dir = parse.path
+
+    def get(self, workflow_id: str, key: str) -> Optional[str]:
+        """
+        Get a key value from the filesystem.
+        """
+        return safe_read_file(os.path.join(self._base_dir, workflow_id, key))
+
+    def set(self, workflow_id: str, key: str, value: Optional[str]) -> None:
+        """
+        Set or clear a key value on the filesystem.
+        """
+        # Make sure the directory we need exists.
+        workflow_dir = os.path.join(self._base_dir, workflow_id)
+        os.makedirs(workflow_dir, exist_ok=True)
+        file_path = os.path.join(workflow_dir, key)
+        if value is None:
+            # Delete the file
+            try:
+                os.unlink(file_path)
+            except FileNotFoundError:
+                # It wasn't there to start with
+                pass
+        else:
+            # Set the value in the file
+            safe_write_file(file_path, value)
+
+if HAVE_S3:
+    class S3StateStore(AbstractStateStore):
+        """
+        A place to store workflow state that uses an S3-compatible object store.
+        """
+
+        def __init__(self, url: str) -> None:
+            """
+            Connect to the state store in the given S3 URL.
+
+            :param url: An S3 URL to a prefix. Interpreted as a URL, so can't
+                        contain ? or #.
+            """
+
+            super().__init__()
+
+            parse = urlparse(url)
+
+            if parse.scheme.lower() != 's3':
+                # We want to catch if we get the wrong argument.
+                raise RuntimeError(f"{url} doesn't look like an S3 URL")
+
+            self._bucket = parse.netloc
+            self._base_path = parse.path
+            self._client = client('s3', region_name=get_current_aws_region())
+
+            logger.debug("Connected to S3StateStore at %s", url)
+
+        def _get_bucket_and_path(self, workflow_id: str, key: str) -> Tuple[str, str]:
+            """
+            Get the bucket and path in the bucket at which a key value belongs.
+            """
+            path = os.path.join(self._base_path, workflow_id, key)
+            return self._bucket, path
+
+        def get(self, workflow_id: str, key: str) -> Optional[str]:
+            """
+            Get a key value from S3.
+            """
+            bucket, path = self._get_bucket_and_path(workflow_id, key)
+            for attempt in retry_s3():
+                try:
+                    logger.debug('Fetch %s path %s', bucket, path)
+                    response = self._client.get_object(Bucket=bucket, Key=path)
+                    return response['Body'].read().decode('utf-8')
+                except self._client.exceptions.NoSuchKey:
+                    return None
+
+
+        def set(self, workflow_id: str, key: str, value: Optional[str]) -> None:
+            """
+            Set or clear a key value on S3.
+            """
+            bucket, path = self._get_bucket_and_path(workflow_id, key)
+            for attempt in retry_s3():
+                if value is None:
+                    # Get rid of it.
+                    logger.debug('Clear %s path %s', bucket, path)
+                    self._client.delete_object(Bucket=bucket, Key=path)
+                    return
+                else:
+                    # Store it, clobbering anything there already.
+                    logger.debug('Set %s path %s', bucket, path)
+                    self._client.put_object(Bucket=bucket, Key=path,
+                                            Body=value.encode('utf-8'))
+                    return
+
+# We want to memoize state stores so we can cache on them.
+state_store_cache: Dict[str, AbstractStateStore] = {}
+
+def connect_to_state_store(url: str) -> AbstractStateStore:
+    """
+    Connect to a place to store state for workflows, defined by a URL.
+
+    URL may be a local file path or URL or an S3 URL.
+    """
+
+    if url not in state_store_cache:
+        # We need to actually make the state store
+        parse = urlparse(url)
+        if parse.scheme.lower() == 's3':
+            # It's an S3 URL
+            if HAVE_S3:
+                # And we can use S3, so make the right implementation for S3.
+                state_store_cache[url] = S3StateStore(url)
+            else:
+                # We can't actually use S3, so complain.
+                raise RuntimeError(f'Cannot connect to {url} because Toil AWS '
+                                   f'dependencies are not available. Did you '
+                                   f'install Toil with the [aws] extra?')
+        elif parse.scheme.lower() in ['file', '']:
+            # It's a file URL or path
+            state_store_cache[url] = FileStateStore(url)
+        else:
+            raise RuntimeError(f'Cannot connect to {url} because we do not '
+                               f'implement its URL scheme')
+
+    return state_store_cache[url]
+
+class WorkflowStateStore:
+    """
+    Slice of a state store for the state of a particular workflow.
+    """
+
+    def __init__(self, state_store: AbstractStateStore, workflow_id: str) -> None:
+        """
+        Wrap the given state store for access to the given workflow's state.
+        """
+
+        # TODO: We could just use functools.partial on the state store methods
+        # to make ours dynamically but that might upset MyPy.
+        self._state_store = state_store
+        self._workflow_id = workflow_id
+
+    def get(self, key: str) -> Optional[str]:
+        """
+        Get the given item of workflow state.
+        """
+        return self._state_store.get(self._workflow_id, key)
+
+    def set(self, key: str, value: Optional[str]) -> None:
+        """
+        Set the given item of workflow state.
+        """
+        self._state_store.set(self._workflow_id, key, value)
+
+    def read_cache(self, key: str) -> Optional[str]:
+        """
+        Read a value from a local cache, without checking the actual backend.
+        """
+
+        return self._state_store.read_cache(self._workflow_id, key)
+
+    def write_cache(self, key: str, value: Optional[str]) -> None:
+        """
+        Write a value to a local cache, without modifying the actual backend.
+        """
+
+        self._state_store.write_cache(self._workflow_id, key, value)
+
+
+def connect_to_workflow_state_store(url: str, workflow_id: str) -> WorkflowStateStore:
+    """
+    Connect to a place to store state for the given workflow, in the state
+    store defined by the given URL.
+
+    :param url: A URL that can be used for connect_to_state_store()
+    """
+
+    return WorkflowStateStore(connect_to_state_store(url), workflow_id)
+
+# When we see one of these terminal states, we stay there forever.
+TERMINAL_STATES = {"COMPLETE", "EXECUTOR_ERROR", "SYSTEM_ERROR", "CANCELED"}
+
+# How long can a workflow be in CANCELING state before we conclude that the
+# workflow running task is gone and move it to CANCELED?
+MAX_CANCELING_SECONDS = 600
+
+class WorkflowStateMachine:
+    """
+    Class for managing the WES workflow state machine.
+
+    This is the authority on the WES "state" of a workflow. You need one to
+    read or change the state.
+
+    Guaranteeing that only certain transitions can be observed is possible but
+    not worth it. Instead, we just let updates clobber each other and grab and
+    cache the first terminal state we see forever. If it becomes important that
+    clients never see e.g. CANCELED -> COMPLETE or COMPLETE -> SYSTEM_ERROR, we
+    can implement a real distributed state machine here.
+
+    We do handle making sure that tasks don't get stuck in CANCELING.
+
+    State can be:
+
+    "UNKNOWN"
+    "QUEUED"
+    "INITIALIZING"
+    "RUNNING"
+    "PAUSED"
+    "COMPLETE"
+    "EXECUTOR_ERROR"
+    "SYSTEM_ERROR"
+    "CANCELED"
+    "CANCELING"
+
+    Uses the state store's local cache to prevent needing to read things we've
+    seen already.
+    """
+
+    def __init__(self, store: WorkflowStateStore) -> None:
+        """
+        Make a new state machine over the given state store slice for the
+        workflow.
+        """
+        self._store = store
+
+    def _set_state(self, state: str) -> None:
+        """
+        Set the state to the given value, if a read does not show a terminal
+        state already.
+        We still might miss and clobber transitions to terminal states between
+        the read and the write.
+        This is not really consistent but also not worth protecting against.
+        """
+
+        if self.get_current_state() not in TERMINAL_STATES:
+            self._store.set("state", state)
+
+    def send_enqueue(self) -> None:
+        """
+        Send an enqueue message that would move from UNKNOWN to QUEUED.
+        """
+        self._set_state("QUEUED")
+
+    def send_initialize(self) -> None:
+        """
+        Send an initialize message that would move from QUEUED to INITIALIZING.
+        """
+        self._set_state("INITIALIZING")
+
+    def send_run(self) -> None:
+        """
+        Send a run message that would move from INITIALIZING to RUNNING.
+        """
+        self._set_state("RUNNING")
+
+    def send_cancel(self) -> None:
+        """
+        Send a cancel message that would move to CANCELING from any
+        non-terminal state.
+        """
+
+        state = self.get_current_state()
+        if state != "CANCELING" and state not in TERMINAL_STATES:
+            # If it's not obvious we shouldn't cancel, cancel.
+
+            # If we end up in CANCELING but the workflow runner task isn't around,
+            # or we signal it at the wrong time, we will stay there forever,
+            # because it's responsible for setting the state to anything else.
+            # So, we save a timestamp, and if we see a CANCELING status and an old
+            # timestamp, we move on.
+            self._store.set("cancel_time", get_iso_time())
+            # Set state after time, because having the state but no time is an error.
+            self._store.set("state", "CANCELING")
+
+    def send_canceled(self) -> None:
+        """
+        Send a canceled message that would move to CANCELED from CANCELLING.
+        """
+        self._set_state("CANCELED")
+
+    def send_complete(self) -> None:
+        """
+        Send a complete message that would move from RUNNING to COMPLETE.
+        """
+        self._set_state("COMPLETE")
+
+    def send_executor_error(self) -> None:
+        """
+        Send an executor_error message that would move from QUEUED,
+        INITIALIZING, or RUNNING to EXECUTOR_ERROR.
+        """
+        self._set_state("EXECUTOR_ERROR")
+
+    def send_system_error(self) -> None:
+        """
+        Send a system_error message that would move from QUEUED, INITIALIZING,
+            or RUNNING to SYSTEM_ERROR.
+        """
+        self._set_state("SYSTEM_ERROR")
+
+    def get_current_state(self) -> str:
+        """
+        Get the current state of the workflow.
+        """
+
+        state = self._store.read_cache("state")
+        if state is not None:
+            # We permanently cached a terminal state
+            return state
+
+        # Otherwise do an actual read from backing storage.
+        state = self._store.get("state")
+
+        if state == "CANCELING":
+            # Make sure it hasn't been CANCELING for too long.
+            # We can get stuck in CANCELING if the workflow-running task goes
+            # away or is stopped while reporting back, because it is
+            # repsonsible for posting back that it has been successfully
+            # canceled.
+            canceled_at = self._store.get("cancel_time")
+            if canceled_at is None:
+                # If there's no timestamp but it's supposedly canceling, put it
+                # into SYSTEM_ERROR, because we didn;t move to CANCELING properly.
+                state = "SYSTEM_ERROR"
+                self._store.set("state", state)
+            else:
+                # See if it has been stuck canceling for too long
+                canceled_at = datetime.fromisoformat(canceled_at)
+                canceling_seconds = (datetime.now() - canceled_at).total_seconds()
+                if canceling_seconds > MAX_CANCELING_SECONDS:
+                    # If it has, go to CANCELED instead, because the task is
+                    # nonresponsive and thus not running.
+                    state = "CANCELED"
+                    self._store.set("state", state)
+
+        if state in TERMINAL_STATES:
+            # We can cache this state forever
+            self._store.write_cache("state", state)
+
+        if state is None:
+            # Make sure we fill in if we couldn't fetch a stored state.
+            state = "UNKNOWN"
+
+        return state
+
+
diff --git a/src/toil/server/wes/tasks.py b/src/toil/server/wes/tasks.py
index c0c190cd..867e3a9a 100644
--- a/src/toil/server/wes/tasks.py
+++ b/src/toil/server/wes/tasks.py
@@ -14,10 +14,11 @@
 import fcntl
 import json
 import logging
+import multiprocessing
 import os
 import subprocess
 import zipfile
-from typing import Dict, Any, List, Optional, Union
+from typing import Dict, Any, List, Optional, Tuple, Union
 from urllib.parse import urldefrag
 
 from celery.exceptions import SoftTimeLimitExceeded  # type: ignore
@@ -29,8 +30,8 @@ from toil.server.utils import (get_iso_time,
                                download_file_from_internet,
                                download_file_from_s3,
                                get_file_class,
-                               safe_read_file,
-                               safe_write_file)
+                               connect_to_workflow_state_store,
+                               WorkflowStateMachine)
 import toil.server.wes.amazon_wes_utils as amazon_wes_utils
 
 logger = logging.getLogger(__name__)
@@ -44,32 +45,52 @@ class ToilWorkflowRunner:
     that command, and collecting the outputs of the resulting workflow run.
     """
 
-    def __init__(self, work_dir: str, request: Dict[str, Any], engine_options: List[str]):
-        self.work_dir = work_dir
+    def __init__(self, base_scratch_dir: str, state_store_url: str, workflow_id: str, request: Dict[str, Any], engine_options: List[str]):
+        """
+        Make a new ToilWorkflowRunner to actually run a workflow leader based
+        on a WES request.
+
+        :param base_scratch_dir: Base work directory. Workflow scratch directory
+               will be in here under the workflow ID.
+        :param state_store_url: URL to the state store through which we will
+               communicate about workflow state with the WES server.
+        :param workflow_id: ID of the workflow run.
+        :param request: WES request information.
+        :param engine_options: Extra options to pass to Toil.
+        """
+
+        # Find the scratch directory inside the base scratch directory
+        self.scratch_dir = os.path.join(base_scratch_dir, workflow_id)
+
+        # Connect to the workflow state store
+        self.store = connect_to_workflow_state_store(state_store_url, workflow_id)
+        # And use a state machine over that to look at workflow state
+        self.state_machine = WorkflowStateMachine(self.store)
+
         self.request = request
         self.engine_options = engine_options
 
         self.wf_type: str = request["workflow_type"].lower().strip()
         self.version: str = request["workflow_type_version"]
 
-        self.exec_dir = os.path.join(self.work_dir, "execution")
-        self.out_dir = os.path.join(self.work_dir, "outputs")
+        self.exec_dir = os.path.join(self.scratch_dir, "execution")
+        self.out_dir = os.path.join(self.scratch_dir, "outputs")
 
         # Compose the right kind of job store to use it the user doesn't specify one.
         default_type = os.getenv('TOIL_WES_JOB_STORE_TYPE', 'file')
-        self.default_job_store = generate_locator(default_type, local_suggestion=os.path.join(self.work_dir, "toil_job_store"))
+        self.default_job_store = generate_locator(default_type, local_suggestion=os.path.join(self.scratch_dir, "toil_job_store"))
 
         self.job_store = self.default_job_store
 
-    def write(self, filename: str, contents: str) -> None:
-        with open(os.path.join(self.work_dir, filename), "w") as f:
+    def write_scratch_file(self, filename: str, contents: str) -> None:
+        """
+        Write a file to the scratch directory.
+        """
+        with open(os.path.join(self.scratch_dir, filename), "w") as f:
             f.write(contents)
 
     def get_state(self) -> str:
-        return safe_read_file(os.path.join(self.work_dir, "state")) or "UNKNOWN"
-
-    def set_state(self, state: str) -> None:
-        safe_write_file(os.path.join(self.work_dir, "state"), state)
+        return self.state_machine.get_current_state()
 
     def write_workflow(self, src_url: str) -> str:
         """
@@ -265,8 +286,8 @@ class ToilWorkflowRunner:
         Calls a command with Popen. Writes stdout, stderr, and the command to
         separate files.
         """
-        stdout_f = os.path.join(self.work_dir, "stdout")
-        stderr_f = os.path.join(self.work_dir, "stderr")
+        stdout_f = os.path.join(self.scratch_dir, "stdout")
+        stderr_f = os.path.join(self.scratch_dir, "stderr")
 
         with open(stdout_f, "w") as stdout, open(stderr_f, "w") as stderr:
             logger.info(f"Calling: '{' '.join(cmd)}'")
@@ -281,35 +302,24 @@ class ToilWorkflowRunner:
         """
 
         # the task has been picked up by the runner and is currently preparing to run
-        self.set_state("INITIALIZING")
+        self.state_machine.send_initialize()
         commands = self.initialize_run()
 
         # store the job store location
-        with open(os.path.join(self.work_dir, "job_store"), "w") as f:
-            f.write(self.job_store)
+        self.store.set("job_store", self.job_store)
 
-        # lock the state file until we start the subprocess
-        file_obj = open(os.path.join(self.work_dir, "state"), "r+")
-        fcntl.flock(file_obj.fileno(), fcntl.LOCK_EX)
-
-        state = file_obj.read()
+        # Check if we are supposed to cancel
+        state = self.get_state()
         if state in ("CANCELING", "CANCELED"):
             logger.info("Workflow canceled.")
             return
 
-        # https://stackoverflow.com/a/15976014
-        file_obj.seek(0)
-        file_obj.write("RUNNING")
-        file_obj.truncate()
-
+        # Otherwise start to run
+        self.state_machine.send_run()
         process = self.call_cmd(cmd=commands, cwd=self.exec_dir)
 
-        # Now the command is running, we can allow state changes again
-        fcntl.flock(file_obj.fileno(), fcntl.LOCK_UN)
-        file_obj.close()
-
-        self.write("start_time", get_iso_time())
-        self.write("cmd", " ".join(commands))
+        self.store.set("start_time", get_iso_time())
+        self.store.set("cmd", " ".join(commands))
 
         try:
             exit_code = process.wait()
@@ -323,16 +333,16 @@ class ToilWorkflowRunner:
             logger.info("Child process terminated by interruption.")
             exit_code = 130
 
-        self.write("end_time", get_iso_time())
-        self.write("exit_code", str(exit_code))
+        self.store.set("end_time", get_iso_time())
+        self.store.set("exit_code", str(exit_code))
 
         if exit_code == 0:
-            self.set_state("COMPLETE")
+            self.state_machine.send_complete()
         # non-zero exit code indicates failure
         elif exit_code == 130:
-            self.set_state("CANCELED")
+            self.state_machine.send_canceled()
         else:
-            self.set_state("EXECUTOR_ERROR")
+            self.state_machine.send_executor_error()
 
     def write_output_files(self) -> None:
         """
@@ -345,7 +355,7 @@ class ToilWorkflowRunner:
         # For CWL workflows, the stdout should be a JSON object containing the outputs
         if self.wf_type == "cwl":
             try:
-                with open(os.path.join(self.work_dir, "stdout")) as f:
+                with open(os.path.join(self.scratch_dir, "stdout")) as f:
                     output_obj = json.load(f)
             except Exception as e:
                 logger.warning("Failed to read outputs object from stdout:", exc_info=e)
@@ -361,15 +371,21 @@ class ToilWorkflowRunner:
 
         # TODO: fetch files from other job stores
 
-        self.write("outputs.json", json.dumps(output_obj))
-
+        self.write_scratch_file("outputs.json", json.dumps(output_obj))
 
-@celery.task(name="run_wes") # type: ignore
-def run_wes(work_dir: str, request: Dict[str, Any], engine_options: List[str]) -> str:
+def run_wes_task(base_scratch_dir: str, state_store_url: str, workflow_id: str, request: Dict[str, Any], engine_options: List[str]) -> str:
     """
-    A celery task to run a requested workflow.
+    Run a requested workflow.
+
+    :param base_scratch_dir: Directory where the workflow's scratch dir will live, under the workflow's ID.
+
+    :param state_store_url: URL/path at which the server and Celery task communicate about workflow state.
+
+    :param workflow_id: ID of the workflow run.
     """
-    runner = ToilWorkflowRunner(work_dir, request=request, engine_options=engine_options)
+    
+    runner = ToilWorkflowRunner(base_scratch_dir, state_store_url, workflow_id,
+                                request=request, engine_options=engine_options)
 
     try:
         runner.run()
@@ -381,16 +397,72 @@ def run_wes(work_dir: str, request: Dict[str, Any], engine_options: List[str]) -
             logger.info(f"Fetching output files.")
             runner.write_output_files()
     except (KeyboardInterrupt, SystemExit, SoftTimeLimitExceeded):
-        runner.set_state("CANCELED")
+        # We canceled the workflow run
+        runner.state_machine.send_canceled()
     except Exception as e:
-        runner.set_state("EXECUTOR_ERROR")
+        # The workflow run broke. We still count as the executor here.
+        runner.state_machine.send_executor_error()
         raise e
 
     return runner.get_state()
 
+# Wrap the task function as a Celery task
+run_wes = celery.task(name="run_wes")(run_wes_task)
 
 def cancel_run(task_id: str) -> None:
     """
     Send a SIGTERM signal to the process that is running task_id.
     """
     celery.control.terminate(task_id, signal='SIGUSR1')
+
+class TaskRunner:
+    """
+    Abstraction over the Celery API. Runs our run_wes task and allows canceling it.
+    
+    We can swap this out in the server to allow testing without Celery.
+    """
+    
+    @staticmethod
+    def run(args: Tuple[str, str, str, Dict[str, Any], List[str]], task_id: str) -> None:
+        """
+        Run the given task args with the given ID on Celery.
+        """
+        run_wes.apply_async(args=args,
+                            task_id=task_id,
+                            ignore_result=True)
+                            
+    @staticmethod
+    def cancel(task_id: str) -> None:
+        """
+        Cancel the task with the given ID on Celery.
+        """
+        cancel_run(task_id)
+        
+
+# If Celery can't be set up, we can just use this fake version instead.
+
+_id_to_process = {}
+class MultiprocessingTaskRunner(TaskRunner):
+    """
+    Version of TaskRunner that just runs tasks with Multiprocessing.
+    """
+    
+    @staticmethod
+    def run(args: Tuple[str, str, str, Dict[str, Any], List[str]], task_id: str) -> None:
+        """
+        Run the given task args with the given ID.
+        """
+        logger.info("Starting task %s in a process", task_id)
+        _id_to_process[task_id] = multiprocessing.Process(target=run_wes_task, args=args)
+        _id_to_process[task_id].start()
+        
+    @staticmethod
+    def cancel(task_id: str) -> None:
+        """
+        Cancel the task with the given ID.
+        """
+        if task_id in _id_to_process:
+            logger.info("Stopping process for task %s", task_id)
+            _id_to_process[task_id].terminate()
+        else:
+            logger.error("Tried to kill nonexistent task %s", task_id)
diff --git a/src/toil/server/wes/toil_backend.py b/src/toil/server/wes/toil_backend.py
index 92c0a330..f10fae6d 100644
--- a/src/toil/server/wes/toil_backend.py
+++ b/src/toil/server/wes/toil_backend.py
@@ -17,8 +17,9 @@ import os
 import shutil
 import uuid
 from collections import Counter
+from contextlib import contextmanager
 from tempfile import NamedTemporaryFile
-from typing import Optional, List, Dict, Any, overload, Generator, Tuple
+from typing import Optional, List, Dict, Any, overload, Generator, TextIO, Tuple, Type
 
 from flask import send_from_directory
 from flask.globals import request as flask_request
@@ -26,7 +27,7 @@ from werkzeug.utils import redirect
 from werkzeug.wrappers.response import Response
 
 
-from toil.server.utils import safe_read_file, safe_write_file
+from toil.server.utils import WorkflowStateMachine, connect_to_workflow_state_store
 from toil.server.wes.abstract_backend import (WESBackend,
                                               handle_errors,
                                               WorkflowNotFoundException,
@@ -34,8 +35,10 @@ from toil.server.wes.abstract_backend import (WESBackend,
                                               VersionNotImplementedException,
                                               WorkflowExecutionException,
                                               OperationForbidden)
-from toil.server.wes.tasks import run_wes, cancel_run
+from toil.server.wes.tasks import TaskRunner, MultiprocessingTaskRunner
 
+from toil.lib.threading import global_mutex
+from toil.lib.io import AtomicFileCreate
 from toil.version import baseVersion
 
 logger = logging.getLogger(__name__)
@@ -43,103 +46,136 @@ logging.basicConfig(level=logging.INFO)
 
 
 class ToilWorkflow:
-    def __init__(self, run_id: str, work_dir: str):
+    def __init__(self, base_scratch_dir: str, state_store_url: str, run_id: str, ):
         """
         Class to represent a Toil workflow. This class is responsible for
-        launching workflow runs and retrieving data generated from them.
+        launching workflow runs via Celery and retrieving data generated from
+        them.
 
-        :param run_id: A uuid string.  Used to name the folder that contains
+        :param base_scratch_dir: The directory where workflows keep their
+                                 output/scratch directories under their run
+                                 IDs.
+
+        :param state_store_url: URL or file path at which we communicate with
+                                running workflows.
+
+        :param run_id: A unique per-run string.  Used to name the folder that contains
                        all of the files containing this particular workflow
                        instance's information.
-        :param work_dir: The parent working directory.
         """
+        self.base_scratch_dir = base_scratch_dir
+        self.state_store_url = state_store_url
         self.run_id = run_id
-        self.work_dir = work_dir
-        self.exec_dir = os.path.join(self.work_dir, "execution")
+
+        self.scratch_dir = os.path.join(self.base_scratch_dir, self.run_id)
+        self.exec_dir = os.path.join(self.scratch_dir, "execution")
+
+        # TODO: share a base class with ToilWorkflowRunner for some of this stuff?
+
+        # Connect to the workflow state store
+        self.store = connect_to_workflow_state_store(state_store_url, self.run_id)
+        # And use a state machine over that to look at workflow state
+        self.state_machine = WorkflowStateMachine(self.store)
 
     @overload
-    def fetch(self, filename: str, default: str) -> str: ...
+    def fetch_state(self, key: str, default: str) -> str: ...
     @overload
-    def fetch(self, filename: str, default: None = None) -> Optional[str]: ...
+    def fetch_state(self, key: str, default: None = None) -> Optional[str]: ...
 
-    def fetch(self, filename: str, default: Optional[str] = None) -> Optional[str]:
+    def fetch_state(self, key: str, default: Optional[str] = None) -> Optional[str]:
         """
-        Return the contents of the given file. If the file does not exist, the
-        default value is returned.
+        Return the contents of the given key in the workflow's state
+        store. If the key does not exist, the default value is returned.
         """
-        if os.path.exists(os.path.join(self.work_dir, filename)):
-            with open(os.path.join(self.work_dir, filename), "r") as f:
-                return f.read()
-        return default
+        value = self.store.get(key)
+        if value is None:
+            return default
+        return value
+
+    @contextmanager
+    def fetch_scratch(self, filename: str) -> Generator[Optional[TextIO], None, None]:
+        """
+        Get a context manager for either a stream for the given file from the
+        workflow's scratch directory, or None if it isn't there.
+        """
+        if os.path.exists(os.path.join(self.scratch_dir, filename)):
+            with open(os.path.join(self.scratch_dir, filename), "r") as f:
+                yield f
+        else:
+            yield None
 
     def exists(self) -> bool:
         """ Return True if the workflow run exists."""
-        return os.path.isdir(self.work_dir)
+        return self.get_state() != "UNKNOWN"
 
     def get_state(self) -> str:
         """ Return the state of the current run."""
-        return safe_read_file(os.path.join(self.work_dir, "state")) or "UNKNOWN"
-
-    def set_state(self, state: str) -> None:
-        """ Set the state for the current run."""
-        safe_write_file(os.path.join(self.work_dir, "state"), state)
+        return self.state_machine.get_current_state()
 
     def set_up_run(self) -> None:
         """ Set up necessary directories for the run."""
-        if not os.path.exists(self.exec_dir):
-            os.makedirs(self.exec_dir)
+        # Go to queued state
+        self.state_machine.send_enqueue()
 
-        # create the state file atomically
-        with NamedTemporaryFile(mode='w', dir=self.work_dir, prefix='state.', delete=False) as f:
-            f.write("QUEUED")
-        os.rename(f.name, os.path.join(self.work_dir, "state"))
+        # Make sure scratch and exec directories exist
+        os.makedirs(self.exec_dir, exist_ok=True)
 
     def clean_up(self) -> None:
         """ Clean directory and files related to the run."""
-        shutil.rmtree(os.path.join(self.work_dir))
+        shutil.rmtree(self.scratch_dir)
+        # Don't remove state; state needs to persist forever.
 
-    def queue_run(self, request: Dict[str, Any], options: List[str]) -> None:
-        """This workflow should be ready to run. Hand this to Celery."""
-        with open(os.path.join(self.work_dir, "request.json"), "w") as f:
+    def queue_run(self, task_runner: Type[TaskRunner], request: Dict[str, Any], options: List[str]) -> None:
+        """This workflow should be ready to run. Hand this to the task system."""
+        with open(os.path.join(self.scratch_dir, "request.json"), "w") as f:
+            # Save the request to disk for get_run_log()
             json.dump(request, f)
 
         try:
-            run_wes.apply_async(args=(self.work_dir, request, options),
-                                task_id=self.run_id,  # set the Celery task ID the same as our run ID
-                                ignore_result=True)
+            # Run the task. Set the task ID the same as our run ID
+            task_runner.run(args=(self.base_scratch_dir, self.state_store_url, self.run_id, request, options),
+                            task_id=self.run_id)
         except Exception:
             # Celery or the broker might be down
-            self.set_state("SYSTEM_ERROR")
+            self.state_machine.send_system_error()
             raise WorkflowExecutionException(f"Failed to run: internal server error.")
 
     def get_output_files(self) -> Any:
         """
         Return a collection of output files that this workflow generated.
         """
-        return json.loads(self.fetch("outputs.json", "{}"))
-
+        with self.fetch_scratch("outputs.json") as f:
+            if f is None:
+                # No file is there
+                return {}
+            else:
+                # Stream in the file
+                return json.load(f)
 
 class ToilBackend(WESBackend):
     """
     WES backend implemented for Toil to run CWL, WDL, or Toil workflows. This
     class is responsible for validating and executing submitted workflows.
-
-    Single machine implementation -
-    Use Celery as the task queue and interact with the "workflows/" directory
-    in the filesystem to store and retrieve data associated with the runs.
     """
 
-    def __init__(self, work_dir: str, options: List[str], dest_bucket_base: Optional[str]) -> None:
+    def __init__(self, work_dir: str, state_store: Optional[str], options: List[str],
+                 dest_bucket_base: Optional[str], bypass_celery: bool = False) -> None:
         """
         Make a new ToilBackend for serving WES.
 
         :param work_dir: Directory to download and run workflows in.
 
+        :param state_store: Path or URL to store workflow state at.
+
         :param options: Command-line options to pass along to workflows. Must
-        use = syntax to set values instead of ordering.
+               use = syntax to set values instead of ordering.
 
         :param dest_bucket_base: If specified, direct CWL workflows to use
-        paths under the given URL for storing output files.
+               paths under the given URL for storing output files.
+
+        :param bypass_celery: Can be set to True to bypass Celery and the
+               message broker and invoke workflow-running tasks without them.
+
         """
         for opt in options:
             if not opt.startswith('-'):
@@ -147,8 +183,76 @@ class ToilBackend(WESBackend):
                 # that would need to remain in the same order.
                 raise ValueError(f'Option {opt} does not begin with -')
         super(ToilBackend, self).__init__(options)
-        self.work_dir = os.path.abspath(work_dir)
+
+        # How should we generate run IDs? We apply a prefix so that we can tell
+        # what things in our work directory suggest that runs exist and what
+        # things don't.
+        self.run_id_prefix = 'run-'
+
+        # Use this to run Celery tasks so we can swap it out for testing.
+        self.task_runner = TaskRunner if not bypass_celery else MultiprocessingTaskRunner
+
         self.dest_bucket_base = dest_bucket_base
+        self.work_dir = os.path.abspath(work_dir)
+        os.makedirs(self.work_dir, exist_ok=True)
+
+        # Where should we talk to the tasks about workflow state?
+
+        if state_store is None:
+            # Store workflow metadata under the work_dir.
+            self.state_store_url = os.path.join(self.work_dir, 'state_store')
+        else:
+            # Use the provided value
+            self.state_store_url = state_store
+
+        # Determine a server identity, so we can guess if a workflow in the
+        # possibly-persistent state store is QUEUED, INITIALIZING, or RUNNING
+        # on a Celery that no longer exists. Ideally we would ask Celery what
+        # Celery cluster it is, or we would reconcile with the tasks that exist
+        # in the Celery cluster, but Celery doesn't seem to have a cluster
+        # identity and doesn't let you poll for task existence:
+        # <https://stackoverflow.com/questions/9824172>
+
+        # Grab an ID for the current kernel boot, if we happen to be Linux.
+        # TODO: Deal with multiple servers in front of the same state store and
+        # file system but on different machines?
+        boot_id = None
+        boot_id_file = "/proc/sys/kernel/random/boot_id"
+        if os.path.exists(boot_id_file):
+            try:
+                with open(boot_id_file) as f:
+                    boot_id = f.readline().strip()
+            except OSError:
+                pass
+        # Assign an ID to the work directory storage.
+        work_dir_id = None
+        work_dir_id_file = os.path.join(self.work_dir, 'id.txt')
+        if os.path.exists(work_dir_id_file):
+            # An ID is assigned already
+            with open(work_dir_id_file) as f:
+                work_dir_id = uuid.UUID(f.readline().strip())
+        else:
+            # We need to try and assign an ID.
+            with global_mutex(self.work_dir, 'id-assignment'):
+                # We need to synchronize with other processes starting up to
+                # make sure we agree on an ID.
+                if os.path.exists(work_dir_id_file):
+                    # An ID is assigned already
+                    with open(work_dir_id_file) as f:
+                        work_dir_id = uuid.UUID(f.readline().strip())
+                else:
+                    work_dir_id = uuid.uuid4()
+                    with AtomicFileCreate(work_dir_id_file) as temp_file:
+                        # Still need to be atomic here or people not locking
+                        # will see an incomplete file.
+                        with open(temp_file, 'w') as f:
+                            f.write(str(work_dir_id))
+        # Now combine into one ID
+        if boot_id is not None:
+            self.server_id = str(uuid.uuid5(work_dir_id, boot_id))
+        else:
+            self.server_id = str(work_dir_id)
+
 
         self.supported_versions = {
             "py": ["3.6", "3.7", "3.8", "3.9"],
@@ -164,13 +268,31 @@ class ToilBackend(WESBackend):
         :param should_exists: If set, ensures that the workflow run exists (or
                               does not exist) according to the value.
         """
-        run = ToilWorkflow(run_id, work_dir=os.path.join(self.work_dir, run_id))
+        run = ToilWorkflow(self.work_dir, self.state_store_url, run_id)
 
         if should_exists and not run.exists():
             raise WorkflowNotFoundException
         if should_exists is False and run.exists():
             raise WorkflowConflictException(run_id)
 
+        # Do a little fixup of orphaned/Martian workflows that were running
+        # before the server bounced and can't be running now.
+        # Sadly we can't just ask Celery if it has heard of them.
+        # TODO: Implement multiple servers working together.
+        owning_server = run.fetch_state("server_id")
+        apparent_state = run.get_state()
+        if (apparent_state not in ("UNKNOWN", "COMPLETE", "EXECUTOR_ERROR", "SYSTEM_ERROR", "CANCELED") and
+            owning_server != self.server_id):
+
+            # This workflow is in a state that suggests it is doing something
+            # but it appears to belong to a previous incarnation of the server,
+            # and so its Celery is probably gone. Put it into system error
+            # state if possible.
+            logger.warning("Run %s in state %s appears to belong to server %s and not us, server %s. "
+                           "Its server is probably gone. Failing the workflow!",
+                           run_id, apparent_state, owning_server, self.server_id)
+            run.state_machine.send_system_error()
+
         return run
 
     def get_runs(self) -> Generator[Tuple[str, str], None, None]:
@@ -179,6 +301,8 @@ class ToilBackend(WESBackend):
             return
 
         for run_id in os.listdir(self.work_dir):
+            if not run_id.startswith(self.run_id_prefix):
+                continue
             run = self._get_run(run_id)
             if run.exists():
                 yield run_id, run.get_state()
@@ -244,10 +368,11 @@ class ToilBackend(WESBackend):
     @handle_errors
     def run_workflow(self) -> Dict[str, str]:
         """ Run a workflow."""
-        run_id = uuid.uuid4().hex
+        run_id = self.run_id_prefix + uuid.uuid4().hex
         run = self._get_run(run_id, should_exists=False)
 
-        # set up necessary directories for the run
+        # set up necessary directories for the run. We need to do this because
+        # we need to save attachments before we can get ahold of the request.
         run.set_up_run()
 
         # stage the uploaded files to the execution directory, so that we can run the workflow file directly
@@ -272,6 +397,11 @@ class ToilBackend(WESBackend):
             run.clean_up()
             raise VersionNotImplementedException(wf_type, version, supported_versions)
 
+        # Now we are actually going to try and do the run.
+
+        # Claim ownership of the run
+        run.store.set("server_id", self.server_id)
+
         # Generate workflow options
         workflow_options = list(self.options)
         if wf_type == "cwl" and self.dest_bucket_base:
@@ -279,7 +409,7 @@ class ToilBackend(WESBackend):
             workflow_options.append('--destBucket=' + os.path.join(self.dest_bucket_base, run_id))
 
         logger.info(f"Putting workflow {run_id} into the queue. Waiting to be picked up...")
-        run.queue_run(request, options=workflow_options)
+        run.queue_run(self.task_runner, request, options=workflow_options)
 
         return {
             "run_id": run_id
@@ -291,15 +421,19 @@ class ToilBackend(WESBackend):
         run = self._get_run(run_id, should_exists=True)
         state = run.get_state()
 
-        request = json.loads(run.fetch("request.json", "{}"))
+        with run.fetch_scratch("request.json") as f:
+            if f is None:
+                request = {}
+            else:
+                request = json.load(f)
 
-        cmd = run.fetch("cmd", "").split("\n")
-        start_time = run.fetch("start_time")
-        end_time = run.fetch("end_time")
+        cmd = run.fetch_state("cmd", "").split("\n")
+        start_time = run.fetch_state("start_time")
+        end_time = run.fetch_state("end_time")
 
         stdout = ""
         stderr = ""
-        if os.path.isfile(os.path.join(run.work_dir, 'stdout')):
+        if os.path.isfile(os.path.join(run.scratch_dir, 'stdout')):
             # We can't use flask_request.host_url here because that's just the
             # hostname, and we need to work mounted at a proxy hostname *and*
             # path under that hostname. So we need to use a relative URL to the
@@ -307,7 +441,7 @@ class ToilBackend(WESBackend):
             stdout = f"../../../../toil/wes/v1/logs/{run_id}/stdout"
             stderr = f"../../../../toil/wes/v1/logs/{run_id}/stderr"
 
-        exit_code = run.fetch("exit_code")
+        exit_code = run.fetch_state("exit_code")
 
         output_obj = {}
         if state == "COMPLETE":
@@ -334,8 +468,10 @@ class ToilBackend(WESBackend):
     def cancel_run(self, run_id: str) -> Dict[str, str]:
         """ Cancel a running workflow."""
         run = self._get_run(run_id, should_exists=True)
-        state = run.get_state()
 
+        # Do some preflight checks on the current state.
+        # We won't catch all cases where the cancel won't go through, but we can catch some.
+        state = run.get_state()
         if state in ("CANCELING", "CANCELED", "COMPLETE"):
             # We don't need to do anything.
             logger.warning(f"A user is attempting to cancel a workflow in state: '{state}'.")
@@ -343,9 +479,10 @@ class ToilBackend(WESBackend):
             # Something went wrong. Let the user know.
             raise OperationForbidden(f"Workflow is in state: '{state}', which cannot be cancelled.")
         else:
-            # Cancel the workflow in the following states: "QUEUED", "INITIALIZING", "RUNNING".
-            run.set_state("CANCELING")
-            cancel_run(run_id)
+            # Go to canceling state if allowed
+            run.state_machine.send_cancel()
+            # Stop the run task if it is there.
+            self.task_runner.cancel(run_id)
 
         return {
             "run_id": run_id
diff --git a/src/toil/utils/toilServer.py b/src/toil/utils/toilServer.py
index 46e4795c..1277c126 100644
--- a/src/toil/utils/toilServer.py
+++ b/src/toil/utils/toilServer.py
@@ -15,6 +15,8 @@
 import logging
 import sys
 
+from toil.statsAndLogging import add_logging_options, set_logging_from_options
+
 logger = logging.getLogger(__name__)
 
 
@@ -27,6 +29,8 @@ def main() -> None:
         sys.exit(1)
 
     parser = parser_with_server_options()
+    add_logging_options(parser)
     args = parser.parse_args()
+    set_logging_from_options(args)
 
     start_server(args)


Test patch:
diff --git a/src/toil/test/server/serverTest.py b/src/toil/test/server/serverTest.py
index 2ff8fe4e..9e17cbf6 100644
--- a/src/toil/test/server/serverTest.py
+++ b/src/toil/test/server/serverTest.py
@@ -17,9 +17,11 @@ import os
 import textwrap
 import time
 import unittest
+import uuid
 import zipfile
+from abc import abstractmethod
 from io import BytesIO
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
 
 try:
     from flask import Flask
@@ -29,35 +31,227 @@ except ImportError:
     # extra wasn't installed. We'll then skip them all.
     pass
 
-from toil.test import ToilTest, needs_server, needs_celery_broker
+from toil.test import ToilTest, needs_server, needs_celery_broker, needs_aws_s3
 
 logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.INFO)
 
-
 @needs_server
 class ToilServerUtilsTest(ToilTest):
     """
     Tests for the utility functions used by the Toil server.
     """
 
+    def test_workflow_canceling_recovery(self):
+        """
+        Make sure that a workflow in CANCELING state will be recovered to a
+        terminal state eventually even if the workflow runner Celery task goes
+        away without flipping the state.
+        """
+
+        from toil.server.utils import WorkflowStateMachine, WorkflowStateStore, MemoryStateStore
+
+        store = WorkflowStateStore(MemoryStateStore(), "test-workflow")
+
+        state_machine = WorkflowStateMachine(store)
+
+        # Cancel a workflow
+        state_machine.send_cancel()
+        # Make sure it worked.
+        self.assertEqual(state_machine.get_current_state(), "CANCELING")
+
+        # Back-date the time of cancelation to something really old
+        store.set("cancel_time", "2011-11-04 00:05:23.283")
+
+        # Make sure it is now CANCELED due to timeout
+        self.assertEqual(state_machine.get_current_state(), "CANCELED")
+
+class hidden:
+    # Hide abstract tests from the test loader
+
+    @needs_server
+    class AbstractStateStoreTest(ToilTest):
+        """
+        Basic tests for state stores.
+        """
+
+        from toil.server.utils import AbstractStateStore
+
+        @abstractmethod
+        def get_state_store(self) -> AbstractStateStore:
+            """
+            Make a state store to test, on a single fixed URL.
+            """
+
+            raise NotImplementedError()
+
+
+        def test_state_store(self) -> None:
+            """
+            Make sure that the state store under test can store and load keys.
+            """
+
+            store = self.get_state_store()
+
+            # Should start None
+            self.assertEqual(store.get('id1', 'key1'), None)
+
+            # Should hold a value
+            store.set('id1', 'key1', 'value1')
+            self.assertEqual(store.get('id1', 'key1'), 'value1')
+
+            # Should distinguish by ID and key
+            self.assertEqual(store.get('id2', 'key1'), None)
+            self.assertEqual(store.get('id1', 'key2'), None)
+
+            store.set('id2', 'key1', 'value2')
+            store.set('id1', 'key2', 'value3')
+            self.assertEqual(store.get('id1', 'key1'), 'value1')
+            self.assertEqual(store.get('id2', 'key1'), 'value2')
+            self.assertEqual(store.get('id1', 'key2'), 'value3')
+
+            # Should allow replacement
+            store.set('id1', 'key1', 'value4')
+            self.assertEqual(store.get('id1', 'key1'), 'value4')
+            self.assertEqual(store.get('id2', 'key1'), 'value2')
+            self.assertEqual(store.get('id1', 'key2'), 'value3')
+
+            # Should show up in another state store
+            store2 = self.get_state_store()
+            self.assertEqual(store2.get('id1', 'key1'), 'value4')
+            self.assertEqual(store2.get('id2', 'key1'), 'value2')
+            self.assertEqual(store2.get('id1', 'key2'), 'value3')
+
+            # Should allow clearing
+            store.set('id1', 'key1', None)
+            self.assertEqual(store.get('id1', 'key1'), None)
+            self.assertEqual(store.get('id2', 'key1'), 'value2')
+            self.assertEqual(store.get('id1', 'key2'), 'value3')
+
+            store.set('id2', 'key1', None)
+            store.set('id1', 'key2', None)
+            self.assertEqual(store.get('id1', 'key1'), None)
+            self.assertEqual(store.get('id2', 'key1'), None)
+            self.assertEqual(store.get('id1', 'key2'), None)
+
+class FileStateStoreTest(hidden.AbstractStateStoreTest):
+    """
+    Test file-based state storage.
+    """
+
+    from toil.server.utils import AbstractStateStore
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.state_store_dir = self._createTempDir()
+
+    def get_state_store(self) -> AbstractStateStore:
+        """
+        Make a state store to test, on a single fixed local path.
+        """
+
+        from toil.server.utils import FileStateStore
+
+        return FileStateStore(self.state_store_dir)
+
+class FileStateStoreURLTest(hidden.AbstractStateStoreTest):
+    """
+    Test file-based state storage using URLs instead of local paths.
+    """
+
+    from toil.server.utils import AbstractStateStore
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.state_store_dir = 'file://' + self._createTempDir()
+
+    def get_state_store(self) -> AbstractStateStore:
+        """
+        Make a state store to test, on a single fixed URL.
+        """
+
+        from toil.server.utils import FileStateStore
+
+        return FileStateStore(self.state_store_dir)
+
+@needs_aws_s3
+class AWSStateStoreTest(hidden.AbstractStateStoreTest):
+    """
+    Test AWS-based state storage.
+    """
+
+    from toil.server.utils import AbstractStateStore
+    from mypy_boto3_s3 import S3ServiceResource
+    from mypy_boto3_s3.service_resource import Bucket
+
+    region: Optional[str]
+    s3_resource: Optional[S3ServiceResource]
+    bucket: Optional[Bucket]
+    bucket_name: Optional[str]
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        """
+        Set up the class with a single pre-existing AWS bucket for all tests.
+        """
+        super().setUpClass()
+
+        from toil.lib.aws import get_current_aws_region, session
+        from toil.lib.aws.utils import create_s3_bucket
+
+        cls.region = get_current_aws_region()
+        cls.s3_resource = session.resource("s3", region_name=cls.region)
+
+        cls.bucket_name = f"toil-test-{uuid.uuid4()}"
+        cls.bucket = create_s3_bucket(cls.s3_resource, cls.bucket_name, cls.region)
+        cls.bucket.wait_until_exists()
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        from toil.lib.aws.utils import delete_s3_bucket
+        if cls.bucket_name:
+            delete_s3_bucket(cls.s3_resource, cls.bucket_name, cls.region)
+        super().tearDownClass()
+
+    def get_state_store(self) -> AbstractStateStore:
+        """
+        Make a state store to test, on a single fixed URL.
+        """
+
+        from toil.server.utils import S3StateStore
+
+        return S3StateStore('s3://' + self.bucket_name)
+
 
 @needs_server
-class ToilWESServerTest(ToilTest):
+class AbstractToilWESServerTest(ToilTest):
     """
-    Tests for Toil's Workflow Execution Service API support using Flask's
-    builtin test client.
+    Class for server tests that provides a self.app in testing mode.
     """
 
+    def __init__(self, *args, **kwargs):
+        """
+        Set up default settings for test classes based on this one.
+        """
+        super().__init__(*args, **kwargs)
+
+        # Default to the local testing task runner instead of Celery for when
+        # we run workflows.
+        self._server_args = ["--bypass_celery"]
+
     def setUp(self) -> None:
-        super(ToilWESServerTest, self).setUp()
+        super().setUp()
         self.temp_dir = self._createTempDir()
 
         from toil.server.app import create_app, parser_with_server_options
         parser = parser_with_server_options()
-        args = parser.parse_args(["--work_dir", os.path.join(self.temp_dir, "workflows")])
+        args = parser.parse_args(self._server_args + ["--work_dir", os.path.join(self.temp_dir, "workflows")])
+
+        # Make the FlaskApp
+        server_app = create_app(args)
 
-        self.app: Flask = create_app(args).app
+        # Fish out the actual Flask
+        self.app: Flask = server_app.app
         self.app.testing = True
 
         self.example_cwl = textwrap.dedent("""
@@ -76,38 +270,7 @@ class ToilWESServerTest(ToilTest):
             """)
 
     def tearDown(self) -> None:
-        super(ToilWESServerTest, self).tearDown()
-
-    def test_home(self) -> None:
-        """ Test the homepage endpoint."""
-        with self.app.test_client() as client:
-            rv = client.get("/")
-        self.assertEqual(rv.status_code, 302)
-
-    def test_health(self) -> None:
-        """ Test the health check endpoint."""
-        with self.app.test_client() as client:
-            rv = client.get("/engine/v1/status")
-        self.assertEqual(rv.status_code, 200)
-
-    def test_get_service_info(self) -> None:
-        """ Test the GET /service-info endpoint."""
-        with self.app.test_client() as client:
-            rv = client.get("/ga4gh/wes/v1/service-info")
-        self.assertEqual(rv.status_code, 200)
-        service_info = json.loads(rv.data)
-
-        self.assertIn("version", service_info)
-        self.assertIn("workflow_type_versions", service_info)
-        self.assertIn("supported_wes_versions", service_info)
-        self.assertIn("supported_filesystem_protocols", service_info)
-        self.assertIn("workflow_engine_versions", service_info)
-        engine_versions = service_info["workflow_engine_versions"]
-        self.assertIn("toil", engine_versions)
-        self.assertEqual(type(engine_versions["toil"]), str)
-        self.assertIn("default_workflow_engine_parameters", service_info)
-        self.assertIn("system_state_counts", service_info)
-        self.assertIn("tags", service_info)
+        super().tearDown()
 
     def _report_log(self, client: "FlaskClient", run_id: str) -> None:
         """
@@ -171,130 +334,187 @@ class ToilWESServerTest(ToilTest):
             logger.info("Waiting on workflow in state %s", state)
             time.sleep(2)
 
-    @needs_celery_broker
-    def test_run_example_cwl_workflow(self) -> None:
+
+class ToilWESServerBenchTest(AbstractToilWESServerTest):
+    """
+    Tests for Toil's Workflow Execution Service API that don't run workflows.
+    """
+
+    def test_home(self) -> None:
+        """ Test the homepage endpoint."""
+        with self.app.test_client() as client:
+            rv = client.get("/")
+        self.assertEqual(rv.status_code, 302)
+
+    def test_health(self) -> None:
+        """ Test the health check endpoint."""
+        with self.app.test_client() as client:
+            rv = client.get("/engine/v1/status")
+        self.assertEqual(rv.status_code, 200)
+
+    def test_get_service_info(self) -> None:
+        """ Test the GET /service-info endpoint."""
+        with self.app.test_client() as client:
+            rv = client.get("/ga4gh/wes/v1/service-info")
+        self.assertEqual(rv.status_code, 200)
+        service_info = json.loads(rv.data)
+
+        self.assertIn("version", service_info)
+        self.assertIn("workflow_type_versions", service_info)
+        self.assertIn("supported_wes_versions", service_info)
+        self.assertIn("supported_filesystem_protocols", service_info)
+        self.assertIn("workflow_engine_versions", service_info)
+        engine_versions = service_info["workflow_engine_versions"]
+        self.assertIn("toil", engine_versions)
+        self.assertEqual(type(engine_versions["toil"]), str)
+        self.assertIn("default_workflow_engine_parameters", service_info)
+        self.assertIn("system_state_counts", service_info)
+        self.assertIn("tags", service_info)
+
+class ToilWESServerWorkflowTest(AbstractToilWESServerTest):
+    """
+    Tests of the WES server running workflows.
+    """
+
+    def run_zip_workflow(self, zip_path: str, include_message: bool = True) -> None:
         """
-        Test submitting the example CWL workflow to the WES server and getting
-        the run status.
+        We have several zip file tests; this submits a zip file and makes sure it ran OK.
         """
+        self.assertTrue(os.path.exists(zip_path))
+        with self.app.test_client() as client:
+            rv = client.post("/ga4gh/wes/v1/runs", data={
+                "workflow_url": "file://" + zip_path,
+                "workflow_type": "CWL",
+                "workflow_type_version": "v1.0",
+                "workflow_params": json.dumps({"message": "Hello, world!"} if include_message else {})
+            })
+            # workflow is submitted successfully
+            self.assertEqual(rv.status_code, 200)
+            self.assertTrue(rv.is_json)
+            run_id = rv.json.get("run_id")
+            self.assertIsNotNone(run_id)
+
+            # Check status
+            self._wait_for_success(client, run_id)
+
+            # TODO: Make sure that the correct message was output!
+
+    def test_run_workflow_relative_url_no_attachments_fails(self) -> None:
+        """Test run example CWL workflow from relative workflow URL but with no attachments."""
+        with self.app.test_client() as client:
+            rv = client.post("/ga4gh/wes/v1/runs", data={
+                "workflow_url": "example.cwl",
+                "workflow_type": "CWL",
+                "workflow_type_version": "v1.0",
+                "workflow_params": "{}"
+            })
+            self.assertEqual(rv.status_code, 400)
+            self.assertTrue(rv.is_json)
+            self.assertEqual(rv.json.get("msg"), "Relative 'workflow_url' but missing 'workflow_attachment'")
+
+    def test_run_workflow_relative_url(self) -> None:
+        """Test run example CWL workflow from relative workflow URL."""
+        with self.app.test_client() as client:
+            rv = client.post("/ga4gh/wes/v1/runs", data={
+                "workflow_url": "example.cwl",
+                "workflow_type": "CWL",
+                "workflow_type_version": "v1.0",
+                "workflow_params": json.dumps({"message": "Hello, world!"}),
+                "workflow_attachment": [
+                    (BytesIO(self.example_cwl.encode()), "example.cwl"),
+                ],
+            })
+            # workflow is submitted successfully
+            self.assertEqual(rv.status_code, 200)
+            self.assertTrue(rv.is_json)
+            run_id = rv.json.get("run_id")
+            self.assertIsNotNone(run_id)
+
+            # Check status
+            self._wait_for_success(client, run_id)
+
+    def test_run_workflow_https_url(self) -> None:
+        """Test run example CWL workflow from the Internet."""
+        with self.app.test_client() as client:
+            rv = client.post("/ga4gh/wes/v1/runs", data={
+                "workflow_url": "https://raw.githubusercontent.com/DataBiosphere/toil/releases/5.4.x/src/toil"
+                                "/test/docs/scripts/cwlExampleFiles/hello.cwl",
+                "workflow_type": "CWL",
+                "workflow_type_version": "v1.0",
+                "workflow_params": json.dumps({"message": "Hello, world!"}),
+            })
+            # workflow is submitted successfully
+            self.assertEqual(rv.status_code, 200)
+            self.assertTrue(rv.is_json)
+            run_id = rv.json.get("run_id")
+            self.assertIsNotNone(run_id)
+
+            # Check status
+            self._wait_for_success(client, run_id)
+
+    def test_run_workflow_single_file_zip(self) -> None:
+        """Test run example CWL workflow from single-file ZIP."""
+        workdir = self._createTempDir()
+        zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
+        with zipfile.ZipFile(zip_path, 'w') as zip_file:
+            zip_file.writestr('example.cwl', self.example_cwl)
+        self.run_zip_workflow(zip_path)
+
+    def test_run_workflow_multi_file_zip(self) -> None:
+        """Test run example CWL workflow from multi-file ZIP."""
+        workdir = self._createTempDir()
+        zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
+        with zipfile.ZipFile(zip_path, 'w') as zip_file:
+            zip_file.writestr('main.cwl', self.example_cwl)
+            zip_file.writestr('distraction.cwl', "Don't mind me")
+        self.run_zip_workflow(zip_path)
+
+    def test_run_workflow_manifest_zip(self) -> None:
+        """Test run example CWL workflow from ZIP with manifest."""
+        workdir = self._createTempDir()
+        zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
+        with zipfile.ZipFile(zip_path, 'w') as zip_file:
+            zip_file.writestr('actual.cwl', self.example_cwl)
+            zip_file.writestr('distraction.cwl', self.example_cwl)
+            zip_file.writestr('MANIFEST.json', json.dumps({"mainWorkflowURL": "actual.cwl"}))
+        self.run_zip_workflow(zip_path)
+
+
+    def test_run_workflow_inputs_zip(self) -> None:
+        """Test run example CWL workflow from ZIP without manifest but with inputs."""
+        workdir = self._createTempDir()
+        zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
+        with zipfile.ZipFile(zip_path, 'w') as zip_file:
+            zip_file.writestr('main.cwl', self.example_cwl)
+            zip_file.writestr('inputs.json', json.dumps({"message": "Hello, world!"}))
+        self.run_zip_workflow(zip_path, include_message=False)
+
+    def test_run_workflow_manifest_and_inputs_zip(self) -> None:
+        """Test run example CWL workflow from ZIP with manifest and inputs."""
+        workdir = self._createTempDir()
+        zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
+        with zipfile.ZipFile(zip_path, 'w') as zip_file:
+            zip_file.writestr('actual.cwl', self.example_cwl)
+            zip_file.writestr('data.json', json.dumps({"message": "Hello, world!"}))
+            zip_file.writestr('MANIFEST.json', json.dumps({"mainWorkflowURL": "actual.cwl", "inputFileURLs": ["data.json"]}))
+        self.run_zip_workflow(zip_path, include_message=False)
+
+    # TODO: When we can check the output value, add tests for overriding
+    # packaged inputs.
+
+@needs_celery_broker
+class ToilWESServerCeleryWorkflowTest(ToilWESServerWorkflowTest):
+    """
+    End-to-end workflow-running tests against Celery.
+    """
+
+    def __init__(self, *args, **kwargs):
+        """
+        Set the task runner back to Celery.
+        """
+        super().__init__(*args, **kwargs)
+        self._server_args = []
 
-        with self.subTest('Test run example CWL workflow from relative workflow URL but with no attachments.'):
-            with self.app.test_client() as client:
-                rv = client.post("/ga4gh/wes/v1/runs", data={
-                    "workflow_url": "example.cwl",
-                    "workflow_type": "CWL",
-                    "workflow_type_version": "v1.0",
-                    "workflow_params": "{}"
-                })
-                self.assertEqual(rv.status_code, 400)
-                self.assertTrue(rv.is_json)
-                self.assertEqual(rv.json.get("msg"), "Relative 'workflow_url' but missing 'workflow_attachment'")
-
-        with self.subTest('Test run example CWL workflow from relative workflow URL.'):
-            with self.app.test_client() as client:
-                rv = client.post("/ga4gh/wes/v1/runs", data={
-                    "workflow_url": "example.cwl",
-                    "workflow_type": "CWL",
-                    "workflow_type_version": "v1.0",
-                    "workflow_params": json.dumps({"message": "Hello, world!"}),
-                    "workflow_attachment": [
-                        (BytesIO(self.example_cwl.encode()), "example.cwl"),
-                    ],
-                })
-                # workflow is submitted successfully
-                self.assertEqual(rv.status_code, 200)
-                self.assertTrue(rv.is_json)
-                run_id = rv.json.get("run_id")
-                self.assertIsNotNone(run_id)
-
-                # Check status
-                self._wait_for_success(client, run_id)
-
-        def run_zip_workflow(zip_path: str, include_message: bool = True) -> None:
-            """
-            We have several zip file tests; this submits a zip file and makes sure it ran OK.
-            """
-            self.assertTrue(os.path.exists(zip_path))
-            with self.app.test_client() as client:
-                rv = client.post("/ga4gh/wes/v1/runs", data={
-                    "workflow_url": "file://" + zip_path,
-                    "workflow_type": "CWL",
-                    "workflow_type_version": "v1.0",
-                    "workflow_params": json.dumps({"message": "Hello, world!"} if include_message else {})
-                })
-                # workflow is submitted successfully
-                self.assertEqual(rv.status_code, 200)
-                self.assertTrue(rv.is_json)
-                run_id = rv.json.get("run_id")
-                self.assertIsNotNone(run_id)
-
-                # Check status
-                self._wait_for_success(client, run_id)
-
-                # TODO: Make sure that the correct message was output!
-
-
-        with self.subTest('Test run example CWL workflow from single-file ZIP.'):
-            workdir = self._createTempDir()
-            zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
-            with zipfile.ZipFile(zip_path, 'w') as zip_file:
-                zip_file.writestr('example.cwl', self.example_cwl)
-            run_zip_workflow(zip_path)
-
-        with self.subTest('Test run example CWL workflow from multi-file ZIP.'):
-            workdir = self._createTempDir()
-            zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
-            with zipfile.ZipFile(zip_path, 'w') as zip_file:
-                zip_file.writestr('main.cwl', self.example_cwl)
-                zip_file.writestr('distraction.cwl', "Don't mind me")
-            run_zip_workflow(zip_path)
-
-        with self.subTest('Test run example CWL workflow from ZIP with manifest.'):
-            workdir = self._createTempDir()
-            zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
-            with zipfile.ZipFile(zip_path, 'w') as zip_file:
-                zip_file.writestr('actual.cwl', self.example_cwl)
-                zip_file.writestr('distraction.cwl', self.example_cwl)
-                zip_file.writestr('MANIFEST.json', json.dumps({"mainWorkflowURL": "actual.cwl"}))
-            run_zip_workflow(zip_path)
-
-        with self.subTest('Test run example CWL workflow from ZIP without manifest but with inputs.'):
-            workdir = self._createTempDir()
-            zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
-            with zipfile.ZipFile(zip_path, 'w') as zip_file:
-                zip_file.writestr('main.cwl', self.example_cwl)
-                zip_file.writestr('inputs.json', json.dumps({"message": "Hello, world!"}))
-            run_zip_workflow(zip_path, include_message=False)
-            
-        with self.subTest('Test run example CWL workflow from ZIP with manifest and inputs.'):
-            workdir = self._createTempDir()
-            zip_path = os.path.abspath(os.path.join(workdir, 'workflow.zip'))
-            with zipfile.ZipFile(zip_path, 'w') as zip_file:
-                zip_file.writestr('actual.cwl', self.example_cwl)
-                zip_file.writestr('data.json', json.dumps({"message": "Hello, world!"}))
-                zip_file.writestr('MANIFEST.json', json.dumps({"mainWorkflowURL": "actual.cwl", "inputFileURLs": ["data.json"]}))
-            run_zip_workflow(zip_path, include_message=False)
-            
-        # TODO: When we can check the output value, add tests for overriding
-        # packaged inputs.
-
-        with self.subTest('Test run example CWL workflow from the Internet.'):
-            with self.app.test_client() as client:
-                rv = client.post("/ga4gh/wes/v1/runs", data={
-                    "workflow_url": "https://raw.githubusercontent.com/DataBiosphere/toil/releases/5.4.x/src/toil"
-                                    "/test/docs/scripts/cwlExampleFiles/hello.cwl",
-                    "workflow_type": "CWL",
-                    "workflow_type_version": "v1.0",
-                    "workflow_params": json.dumps({"message": "Hello, world!"}),
-                })
-                # workflow is submitted successfully
-                self.assertEqual(rv.status_code, 200)
-                self.assertTrue(rv.is_json)
-                run_id = rv.json.get("run_id")
-                self.assertIsNotNone(run_id)
-
-                # Check status
-                self._wait_for_success(client, run_id)
 
 
 if __name__ == "__main__":