UWS conformance: job deletion, 404 for unknown jobs, executionDuration, consistent job values

MIGRATING.md

@@ -17,7 +17,7 @@ line a day to stderr (the web server's error log):
 | `nea-errors` | errors are VOTable documents; a table not in TAP_SCHEMA returns 400 | plain text; 403 |
 | `nea-tables` | `/tables` uses VODataService `<flag>` elements | per-column `<principal>`, `<indexed>`, `<std>`, `<column_index>`, `<arraysize>` |
 | `nea-votable` | VOTable results are `application/xml`, non-char FIELDs carry no DESCRIPTION (units are still written) | `text/xml`, with descriptions on every FIELD |
-| `nea-uws` | job documents use `executionDuration`, `application/xml`, no `Z` on `destruction`; the 303 that starts a job (`PHASE=RUN`) carries no `Content-Type`, `Content-Length` or `Connection` lines, because the response ends when the CGI exits; `/async/<id>/executionDuration` (camelCase) is accepted as well as `/executionduration` | `executionduration`, `text/xml`, `Z`; the 303 carries `Content-Type: text/plain`, `Content-Length` and `Connection: close`; only `/executionduration` is a valid key |
+| `nea-uws` | job documents are `application/xml`, no `Z` on `destruction`; the 303 that starts a job (`PHASE=RUN`) carries no `Content-Type`, `Content-Length` or `Connection` lines, because the response ends when the CGI exits; `/async/<id>/executionDuration` (camelCase) is accepted as well as `/executionduration`; job URLs join `HTTP_URL` and `CGI_PGM` with `/` as written, so `CGI_PGM=/TAP` gives `//TAP` | `text/xml`, `Z`; the 303 carries `Content-Type: text/plain`, `Content-Length` and `Connection: close`; only `/executionduration` is a valid key; job URLs have one `/` between `HTTP_URL` and `CGI_PGM` |
 
 ## Moving to the 3.x layout
 
@@ -48,22 +48,27 @@ Compatibility mode reproduces 1.2's responses, with these exceptions:
 - Errors raised before `TAP.conf` is read (a missing or broken `TAP.conf`, a
   malformed request path) use the 3.x error format, since the mode isn't known
   yet.
-- `/availability` and `/capabilities` create a workspace directory per
-  request, so workspace cleanup must cover them.
 - The async job document escapes quotes in the query text as `&#x27;`; it
   parses to the same XML.
-- A job's `status.xml` written under one mode is read correctly under the
-  other (either spelling of the duration element), so switching `nea-uws` off
-  doesn't break jobs in flight.
+- Both modes write the job document's duration as `executionDuration` (the
+  UWS name) and read either spelling, so jobs written by an earlier release
+  or under the other mode read correctly.
+- Both modes follow UWS where 1.2 did not: a job can be deleted
+  (`POST ACTION=DELETE` or HTTP `DELETE`, answered with 303 to the job list),
+  an unknown job id is 404, `GET` on the job list returns an empty
+  `<uws:jobs>` instead of creating a job, and an unset job value (such as
+  `/destruction` before the job runs) is empty text instead of `None`.
 
 `COMPAT` is read only from `[WEB]`; placing it in any other section, including
 a 1.x `[webserver]`, is a configuration error.
 
 ## Release plan
 
-- **3.1.0**: compatibility mode, the 1.x reader, this guide.
-- **3.x**: where the shared behavior is to change (for example
-  `executionDuration` casing and VOTable errors, which IVOA standards
-  require), the release notes say so before any default changes.
+- **3.1.0**: compatibility mode, the 1.x reader, this guide.  The default
+  job document now spells the duration `executionDuration`, as UWS
+  requires, which retires that part of `nea-uws`.
+- **3.x**: where the shared behavior is to change (for example VOTable
+  errors, which IVOA standards require), the release notes say so before
+  any default changes.
 - **4.0.0**: compatibility mode, the `COMPAT` key and the 1.x reader are
   removed; a 1.x `TAP.conf` then fails at startup with a pointer here.

TAP/compat.py

@@ -169,9 +169,15 @@ def field_description(active, desc, coltype):
 
 # --- nea-uws ----------------------------------------------------------------
 
-def duration_element(active):
-    """UWS names it executionDuration; the shared line writes it lowercase."""
-    return 'executionDuration' if 'nea-uws' in active else 'executionduration'
+def service_url(active, httpurl, cgipgm):
+    """The service URL that job URLs extend.
+
+    NEA's carry a double slash (HTTP_URL + '/' + CGI_PGM=/TAP); the shared
+    line joins the parts with one.
+    """
+    if 'nea-uws' in active:
+        return httpurl + '/' + cgipgm
+    return httpurl.rstrip('/') + '/' + cgipgm.strip('/')
 
 
 def uws_key(job, key):
@@ -199,7 +205,8 @@ def duration_value(job):
     """The duration in a parsed status.xml, whichever spelling wrote it.
 
     A job written under one mode may be read under the other (jobs live up
-    to four days), so reading tolerates both; writing uses duration_element.
+    to four days, and 3.1.0 and earlier wrote it lowercase), so reading
+    tolerates both; writing uses the UWS spelling.
     """
     return job.get('uws:executionDuration', job.get('uws:executionduration'))
 

TAP/tap.py

@@ -9,6 +9,8 @@
 import logging
 import math
 import os
+import re
+import shutil
 import signal
 import sys
 import tempfile
@@ -170,7 +172,13 @@ def __run__(self, **kwargs):
         # { tap.init()
         #
 
-        if('debug' in self.form):
+        # A request with no form fields (an HTTP DELETE) has no list, and
+        # FieldStorage raises 'not indexable' on any lookup into it.
+        self.fields = self.form.keys() if self.form.list is not None else []
+        self.request_method = os.environ.get('REQUEST_METHOD', '').upper()
+        self.action = ''
+
+        if('debug' in self.fields):
             self.debug = 1
 
         if(self.debug):
@@ -227,7 +235,7 @@ def __run__(self, **kwargs):
         self.uwsheader = '<uws:job xmlns:uws="http://www.ivoa.net/xml/UWS/v1.0" xmlns:xlink="http://www.w3.org/1999/xlink" xmlns:xsi="http://www.w3.org/2001/XMLSchema" xsi:schemaLocation="http://www.ivoa.net/xml/UWS/v1.0 http://www.ivoa.net/xml/UWS/v1.0">'
 
 
-        for key in self.form:
+        for key in self.fields:
             if self.debug:
                 logging.debug(f'      key: {key:<15}   val: {self.form[key].value:s}')
 
@@ -267,6 +275,9 @@ def __run__(self, **kwargs):
             if(key.lower() == 'token'):
                 self.token = self.form[key].value
 
+            if(key.lower() == 'action'):
+                self.action = self.form[key].value.strip().upper()
+
         self.nparam = len(self.param)
 
         if 'format' in self.param:
@@ -526,27 +537,6 @@ def __run__(self, **kwargs):
 
 
 
-        #  Special case:  HTTP DELETE.  We only need
-        #  the REQUEST_METHOD and PATH_INFO environment
-        #  variable.  The first must be DELETE and the
-        #  second is the directory of th TAP job we wish
-        #  to delete.
-        #
-
-        # if("REQUEST_METHOD" in os.environ):
-        #     self.request_method = os.environ["REQUEST_METHOD"]
-        # else:
-        #     self.request_method = ''
-        #
-        # if self.request_method == 'DELETE':
-        #
-        #     delete_dir = self.workdir + self.pathinfo
-        #
-        #     if self.debug:
-        #         logging.debug('')
-        #         logging.debug(f'DELETE: {delete_dir:s}')
-
-
         #
         # Initialize statdict dict
         #
@@ -588,6 +578,15 @@ def __run__(self, **kwargs):
         #} end VOSI static endpoints
         #
 
+        #
+        # GET on the job list must not create a job: clients follow the
+        # redirect that answers a job deletion with a GET here.
+        #
+        if ((self.tapcontext == 'async') and (self.getstatus == 0) and
+                (self.querykey == 0) and (len(self.param['phase']) == 0) and
+                (self.request_method != 'POST')):
+            self.__printJobList__()
+
         #
         # sync or async without input workspace id: make workspace,
         # otherwise retrieve workspace from getstatus id
@@ -656,18 +655,34 @@ def __run__(self, **kwargs):
             #   Retrieve workspace from id
             #
 
+            #
+            #    A job id names a directory that tempfile.mkdtemp made under
+            #    <workdir>/TAP; anything else ('..', '.') would reach outside
+            #    it, and a deletion removes what the id names.
+            #
+            if not re.fullmatch(r'tap_[A-Za-z0-9_]+', self.id):
+                self.__printError__('votable', 'Job not found: ' + self.id,
+                                    errcode='404')
+
             self.workspace = self.id
             self.userWorkdir = self.workdir + '/TAP/' + self.workspace
 
 
             #
             #    check if workspace exists
             #
-            isExist = os.path.exists(self.userWorkdir)
+            isExist = os.path.isdir(self.userWorkdir)
 
             if (isExist == 0):
-                self.msg = 'work directory based on input jobid does not exist.'
-                self.__printError__('votable', self.msg, errcode='500')
+                self.__printError__('votable', 'Job not found: ' + self.id,
+                                    errcode='404')
+
+            if ((self.tapcontext == 'async') and
+                    (self.statuskey.strip('/') == '') and
+                    ((self.request_method == 'DELETE') or
+                     ((self.request_method == 'POST') and
+                      (self.action == 'DELETE')))):
+                self.__deleteJob__()
 
             #
             # } end of retrieve workspace
@@ -712,7 +727,8 @@ def __run__(self, **kwargs):
 
         self.statustbl = 'status.xml'
         self.statuspath = self.userWorkdir + '/' + self.statustbl
-        self.statusurl = self.httpurl + '/' + self.cgipgm + \
+        self.statusurl = \
+            compat.service_url(self.compat, self.httpurl, self.cgipgm) + \
             '/' + self.tapcontext + '/' + self.workspace
 
         if self.debug:
@@ -2284,7 +2300,11 @@ def __getStatus__(self, workdir, workspace, key, param, **kwargs):
                     or (key == 'jobId')
                     or (key == 'runId')):
 
-                retval = job[keystr]
+                # xmltodict reads an empty element as None and a nil one
+                # as a dict; both are an unset value, which is empty text.
+                retval = job.get(keystr)
+                if (retval is None) or isinstance(retval, dict):
+                    retval = ''
                 outstr = retval
 
             elif((key == 'ownerId') or (key == 'quote')):
@@ -2756,6 +2776,69 @@ def __printSyncResponse__(self, status, msg, resulturl, format, **kwargs):
         #
 
 
+    def __printJobList__(self, **kwargs):
+
+        #
+        # UWS job list (GET /async).  Jobs carry no owner, so the list
+        # shows none of them rather than everyone's.
+        #
+
+        body = ('<?xml version="1.0" encoding="UTF-8"?>\n'
+                '<uws:jobs xmlns:uws="http://www.ivoa.net/xml/UWS/v1.0"'
+                ' xmlns:xlink="http://www.w3.org/1999/xlink" version="1.1"/>\n')
+
+        sys.stdout.write('HTTP/1.1 200 OK\r\n')
+        sys.stdout.write(f'Content-Type: {compat.uws_content_type(self.compat)}\r\n')
+        sys.stdout.write(f'Content-Length: {len(body.encode("utf-8"))}\r\n')
+        sys.stdout.write('Connection: close\r\n\r\n')
+        sys.stdout.write(body)
+        sys.stdout.flush()
+        sys.exit()
+
+
+    def __deleteJob__(self, **kwargs):
+
+        #
+        # UWS job deletion (POST ACTION=DELETE, or HTTP DELETE): stop the
+        # job if it is running, remove its workspace, and send the client
+        # to the job list.  The job id was checked against the workspace
+        # name pattern before userWorkdir was built from it.
+        #
+
+        try:
+            with open(self.userWorkdir + '/status.xml', 'r') as fp:
+                job = xmltodict.parse(fp.read())['uws:job']
+
+            if (job.get('uws:phase') == 'EXECUTING'):
+                pid = int(job.get('uws:runId') or 0)
+                if (pid > 0) and (pid != os.getpid()):
+                    os.kill(pid, signal.SIGKILL)
+
+        except Exception as e:
+
+            # A job whose status can't be read or whose process is gone
+            # is still deleted.
+            if self.debug:
+                logging.debug(f'deleteJob: {str(e):s}')
+
+        try:
+            shutil.rmtree(self.userWorkdir)
+
+        except OSError as e:
+            self.__printError__('votable', 'Failed to delete job ' +
+                                self.workspace + ': ' + str(e), errcode='500')
+
+        joblist = compat.service_url(self.compat, self.httpurl, self.cgipgm) + \
+            '/async'
+
+        sys.stdout.write('HTTP/1.1 303 See Other\r\n')
+        sys.stdout.write(f'Location: {joblist}\r\n')
+        sys.stdout.write('Content-Length: 0\r\n')
+        sys.stdout.write('Connection: close\r\n\r\n')
+        sys.stdout.flush()
+        sys.exit()
+
+
     def __printAsyncResponse__(self, statusurl, **kwargs):
 
         #
@@ -3051,8 +3134,8 @@ def __writeStatusMsg__(self, statuspath, statdict, param, **kwargs):
 
         fp.write(f"    <uws:endTime>{statdict['endtime']:s}</uws:endTime>\n")
 
-        el = compat.duration_element(self.compat)
-        fp.write(f"    <uws:{el}>{statdict['duration']:d}</uws:{el}>\n")
+        fp.write(f"    <uws:executionDuration>{statdict['duration']:d}"
+                 "</uws:executionDuration>\n")
 
         if (statdict['destruction'] is None):
             fp.write('    <uws:destruction xsi:nil="true"/>\n')

tests/conftest.py

@@ -101,6 +101,9 @@ def do_GET(self):
     def do_POST(self):
         self._run_cgi()
 
+    def do_DELETE(self):
+        self._run_cgi()
+
     def _run_cgi(self):
         # Parse PATH_INFO and QUERY_STRING from self.path.
         from urllib.parse import urlsplit

tests/test_compat_http.py

@@ -119,15 +119,15 @@ def test_uws_is_neas_form(legacy_tap_server):
     assert camel.text == dur.text
 
 
-def test_uws_unchanged_in_default_mode(tap_server):
+def test_uws_in_default_mode(tap_server):
     run, url = _completed_job(tap_server)
     assert run.headers["Content-Type"] == "text/plain"
     assert int(run.headers["Content-Length"]) == len(run.content)
     assert run.headers["Connection"] == "close"
     r = requests.get(url, timeout=30)
     assert r.headers["Content-Type"] == "text/xml"
     job = ET.fromstring(r.text)
-    assert job.find(UWS + "executionduration") is not None
+    assert job.find(UWS + "executionDuration") is not None   # UWS spelling, as of 3.1.0
     assert job.find(UWS + "destruction").text.endswith("Z")
 
 
@@ -140,7 +140,7 @@ def test_camelcase_duration_key_unchanged_in_default_mode(tap_server):
 
 
 def _respell_duration(fixture_root, url, old, new):
-    """Rewrite one job's status.xml as if the other mode had written it."""
+    """Rewrite one job's status.xml as an earlier release would have written it."""
     jobid = url.rstrip("/").rsplit("/", 1)[1]
     paths = list(fixture_root.glob("workdir*/**/%s/status.xml" % jobid))
     assert len(paths) == 1, paths
@@ -158,15 +158,18 @@ def _read_back(url):
     assert phase.text.strip() == "COMPLETED", phase.text[:300]
 
 
-def test_status_written_by_default_mode_reads_in_compat(legacy_tap_server, fixture_root):
+# Jobs live four days, so a job written lowercase by an earlier release can
+# be read after an upgrade, in either mode.
+
+def test_lowercase_status_reads_in_compat(legacy_tap_server, fixture_root):
     _, url = _completed_job(legacy_tap_server)
     _respell_duration(fixture_root, url, "executionDuration", "executionduration")
     _read_back(url)
 
 
-def test_status_written_by_compat_reads_in_default_mode(tap_server, fixture_root):
+def test_lowercase_status_reads_in_default_mode(tap_server, fixture_root):
     _, url = _completed_job(tap_server)
-    _respell_duration(fixture_root, url, "executionduration", "executionDuration")
+    _respell_duration(fixture_root, url, "executionDuration", "executionduration")
     _read_back(url)
 
 

tests/test_compat_units.py

@@ -34,8 +34,6 @@ def test_field_description_blanks_only_non_char_columns():
 
 
 def test_uws_functions():
-    assert compat.duration_element(compat.ALL) == 'executionDuration'
-    assert compat.duration_element(compat.NONE) == 'executionduration'
     assert compat.uws_key({'uws:executionDuration': '1'}, 'executionduration') == 'executionDuration'
     assert compat.uws_key({'uws:executionduration': '1'}, 'executionduration') == 'executionduration'
     assert compat.uws_key({'uws:executionDuration': '1'}, 'phase') == 'phase'