From 6090a36a27667b11dc764a81d1c785ce25cd5586 Mon Sep 17 00:00:00 2001 From: Paul Elliott Date: Mon, 20 Jul 2026 21:02:12 -0400 Subject: [PATCH 01/12] Add VolView processing backend --- MANIFEST.in | 1 + girder_volview/__init__.py | 596 +- girder_volview/backend/__init__.py | 3 + girder_volview/backend/config.py | 57 + girder_volview/backend/inputs.py | 399 ++ girder_volview/backend/launch.py | 510 ++ girder_volview/backend/outputs.py | 401 ++ girder_volview/backend/results.py | 427 ++ girder_volview/backend/routes.py | 918 +++ girder_volview/backend/slicer_spec.py | 816 +++ girder_volview/backend/submit.py | 649 ++ girder_volview/dicom.py | 30 +- girder_volview/handles.py | 71 + girder_volview/utils.py | 454 +- girder_volview/web_client/package-lock.json | 6265 +++++++++++++++++ girder_volview/web_client/package.json | 43 +- .../web_client/views/HierarchyWidget.js | 82 +- girder_volview/web_client/views/itemPage.js | 2 +- girder_volview/web_client/views/open.js | 37 +- girder_volview/web_client/webpack.helper.js | 4 +- session_builder/README.md | 2 + session_builder/composable_example.py | 5 +- session_builder/folder_session_example.py | 2 - session_builder/item_session_example.py | 2 - session_builder/itk_analysis_example.py | 9 +- session_builder/session_builder.py | 33 +- session_builder/totalsegmentator_example.py | 3 +- setup.py | 10 +- tests/conftest.py | 247 + tests/contract_loader.py | 93 + .../expected/masked-median-filter.json | 53 + tests/slicer_xml/expected/median-filter.json | 41 + .../expected/otsu-segmentation.json | 41 + .../expected/synthetic-bounds-enum.json | 65 + .../expected/threshold-segmentation.json | 47 + tests/slicer_xml/masked-median-filter.xml | 68 + tests/slicer_xml/median-filter.xml | 61 + tests/slicer_xml/otsu-segmentation.xml | 61 + tests/slicer_xml/synthetic-bounds-enum.xml | 70 + tests/slicer_xml/threshold-segmentation.xml | 63 + tests/test_config_group_scope.py | 88 + tests/test_container_token_scope.py | 84 + tests/test_contract_fixtures.py | 130 + tests/test_end_to_end_live.py | 372 + tests/test_filter_match.py | 26 +- tests/test_folder_delete_cascade.py | 494 ++ tests/test_handle_roundtrip.py | 133 + tests/test_input_resolution_routes.py | 313 + tests/test_input_value_resolution.py | 584 ++ tests/test_item_launch_parity.py | 309 + tests/test_job_addressed_routes.py | 167 + tests/test_job_deletion_routes.py | 149 + tests/test_job_history_durability.py | 416 ++ tests/test_job_history_durability_routes.py | 548 ++ tests/test_job_output_binding.py | 613 ++ tests/test_job_output_binding_routes.py | 488 ++ tests/test_job_results_conformance.py | 125 + tests/test_legacy_open_through.py | 120 + tests/test_load.py | 4 +- tests/test_openapi_conformance.py | 101 + tests/test_openapi_envelope_conformance.py | 245 + tests/test_proxiable_download_headers.py | 108 + tests/test_result_intent.py | 193 + tests/test_save_load_restore.py | 630 ++ tests/test_slicer_spec_parse_cli.py | 245 + tests/test_slicer_spec_translation.py | 449 ++ tests/test_staging_routes.py | 383 + tests/test_status_conformance.py | 172 + tests/test_submit_param_guards.py | 419 ++ tests/test_task_scoping.py | 125 + tests/test_task_spec_route.py | 132 + tests/test_transient_cleanup.py | 362 + 72 files changed, 20720 insertions(+), 748 deletions(-) create mode 100644 girder_volview/backend/__init__.py create mode 100644 girder_volview/backend/config.py create mode 100644 girder_volview/backend/inputs.py create mode 100644 girder_volview/backend/launch.py create mode 100644 girder_volview/backend/outputs.py create mode 100644 girder_volview/backend/results.py create mode 100644 girder_volview/backend/routes.py create mode 100644 girder_volview/backend/slicer_spec.py create mode 100644 girder_volview/backend/submit.py create mode 100644 girder_volview/handles.py create mode 100644 girder_volview/web_client/package-lock.json create mode 100644 tests/conftest.py create mode 100644 tests/contract_loader.py create mode 100644 tests/slicer_xml/expected/masked-median-filter.json create mode 100644 tests/slicer_xml/expected/median-filter.json create mode 100644 tests/slicer_xml/expected/otsu-segmentation.json create mode 100644 tests/slicer_xml/expected/synthetic-bounds-enum.json create mode 100644 tests/slicer_xml/expected/threshold-segmentation.json create mode 100644 tests/slicer_xml/masked-median-filter.xml create mode 100644 tests/slicer_xml/median-filter.xml create mode 100644 tests/slicer_xml/otsu-segmentation.xml create mode 100644 tests/slicer_xml/synthetic-bounds-enum.xml create mode 100644 tests/slicer_xml/threshold-segmentation.xml create mode 100644 tests/test_config_group_scope.py create mode 100644 tests/test_container_token_scope.py create mode 100644 tests/test_contract_fixtures.py create mode 100644 tests/test_end_to_end_live.py create mode 100644 tests/test_folder_delete_cascade.py create mode 100644 tests/test_handle_roundtrip.py create mode 100644 tests/test_input_resolution_routes.py create mode 100644 tests/test_input_value_resolution.py create mode 100644 tests/test_item_launch_parity.py create mode 100644 tests/test_job_addressed_routes.py create mode 100644 tests/test_job_deletion_routes.py create mode 100644 tests/test_job_history_durability.py create mode 100644 tests/test_job_history_durability_routes.py create mode 100644 tests/test_job_output_binding.py create mode 100644 tests/test_job_output_binding_routes.py create mode 100644 tests/test_job_results_conformance.py create mode 100644 tests/test_legacy_open_through.py create mode 100644 tests/test_openapi_conformance.py create mode 100644 tests/test_openapi_envelope_conformance.py create mode 100644 tests/test_proxiable_download_headers.py create mode 100644 tests/test_result_intent.py create mode 100644 tests/test_save_load_restore.py create mode 100644 tests/test_slicer_spec_parse_cli.py create mode 100644 tests/test_slicer_spec_translation.py create mode 100644 tests/test_staging_routes.py create mode 100644 tests/test_status_conformance.py create mode 100644 tests/test_submit_param_guards.py create mode 100644 tests/test_task_scoping.py create mode 100644 tests/test_task_spec_route.py create mode 100644 tests/test_transient_cleanup.py diff --git a/MANIFEST.in b/MANIFEST.in index 300cc20..0f5cbdf 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -4,5 +4,6 @@ include setup.py graft girder_volview graft docs +prune girder_volview/web_client/node_modules prune test global-exclude *.py[co] *.cmake __pycache__ node_modules diff --git a/girder_volview/__init__.py b/girder_volview/__init__.py index f032bdc..9a04921 100644 --- a/girder_volview/__init__.py +++ b/girder_volview/__init__.py @@ -1,8 +1,6 @@ import cherrypy -import errno -import copy -from girder import plugin, events +from girder import plugin from girder.api.describe import Description, autoDescribeRoute from girder.api import access from girder.api.rest import ( @@ -10,73 +8,38 @@ setResponseHeader, setContentDisposition, ) -from girder.constants import AccessType, TokenScope, SortDir +from girder.constants import AccessType, TokenScope from girder.models.file import File -from girder.models.upload import Upload -from girder.utility import RequestBodyStream -from girder.exceptions import GirderException, RestException - from girder.models.item import Item -from girder.utility import ziputil -# used by get config -import yaml -from girder.models.setting import Setting from girder.models.folder import Folder -from girder import logger -from girder.models.group import Group # server settings (from girder.cfg file probably) for proxiable endpoint below from girder.utility import config from .dicom import setupEventHandlers -from .utils import ( - SESSION_ZIP_EXTENSION, - isSessionItem, - isLoadableImage, - isLoadableFile, - filesToManifest, - singleVolViewZipOrImageFiles, - idStringToIdList, - normalizeLinkedResources, - loadModels, - getFiles, - findNewestSession, - getLinkedResources, - matchesSelectionSet, - getNewestDoc, - getTouchedTime, - getFilteredFiles, - getFilteredSessionFile, - sessionNameFromFilter, +from .backend import addBackendRoutes +from .backend.launch import ( + downloadManifest, + downloadResourceManifest, + getFolderConfigFile, + saveToItem, + saveToFolder, ) - -LARGE_IMAGE_CONFIG_FOLDER = "large_image.config_folder" - - -BASE_CONFIG = { - "io": {"segmentGroupExtension": "seg", "segmentGroupSaveFormat": "nii.gz", "layerExtension": "layer"}, - "disabledViewTypes": ["3D", "Oblique"], - "layouts": { - "Axial Coronal Sagittal": { - "direction": "row", - "items": [ - "axial", - { - "direction": "column", - "items": ["coronal", "sagittal"] - } - ] - }, - "Axial Only": [["axial"]], - }, -} +from .utils import isLoadableImage, isSessionFile def hasLoadableFile(files, user=None): + # Mirror what the launch manifest would actually resolve: a session file + # opens through (restore), and anything else must be a loadable image that + # is not working data (job outputs, transient staged inputs) — otherwise + # the Open-in-VolView button would launch an empty viewer. + itemCache = {} + folderCache = {} for fileEntry in files: - if isLoadableFile(fileEntry[1] if isinstance(fileEntry, tuple) else fileEntry, user): + file = fileEntry[1] if isinstance(fileEntry, tuple) else fileEntry + if isSessionFile(file) or isLoadableImage(file, user, itemCache, folderCache): return True return False @@ -118,192 +81,70 @@ def volViewLoadableFolder(self, folder): # iterator that yields a tuple of (path, file). The aggregation is much # faster, as it only takes one database roundtrip, rather than one per # folder and one per item. - files = Folder().collection.aggregate([ - {"$match": {"_id": folder["_id"]}}, - {"$graphLookup": { - "from": "folder", - "startWith": folder["_id"], - "connectFromField": "_id", - "connectToField": "parentId", - "as": "__children" - }}, - {"$lookup": { - "from": "folder", - "localField": "_id", - "foreignField": "_id", - "as": "__self", - }}, - {"$project": {"__children": {"$concatArrays": [ - "$__self", "$__children", - ]}}}, - {"$unwind": {"path": "$__children"}}, - {"$replaceRoot": {"newRoot": "$__children"}}, - {"$match": Folder().permissionClauses(self.getCurrentUser(), level=AccessType.READ)}, - {"$lookup": { - "from": "item", - "let": {"fid": "$_id"}, - "pipeline": [ - {"$match": {"$expr": {"$eq": ["$$fid", "$folderId"]}}}, - {"$project": {"_id": 1}}, - ], - "as": "__items", - }}, - {"$lookup": { - "from": "file", - "localField": "__items._id", - "foreignField": "itemId", - "as": "__files", - }}, - {"$unwind": "$__files"}, - {"$replaceRoot": {"newRoot": "$__files"}}, - ]) + files = Folder().collection.aggregate( + [ + {"$match": {"_id": folder["_id"]}}, + { + "$graphLookup": { + "from": "folder", + "startWith": folder["_id"], + "connectFromField": "_id", + "connectToField": "parentId", + "as": "__children", + } + }, + { + "$lookup": { + "from": "folder", + "localField": "_id", + "foreignField": "_id", + "as": "__self", + } + }, + { + "$project": { + "__children": { + "$concatArrays": [ + "$__self", + "$__children", + ] + } + } + }, + {"$unwind": {"path": "$__children"}}, + {"$replaceRoot": {"newRoot": "$__children"}}, + { + "$match": Folder().permissionClauses( + self.getCurrentUser(), level=AccessType.READ + ) + }, + { + "$lookup": { + "from": "item", + "let": {"fid": "$_id"}, + "pipeline": [ + {"$match": {"$expr": {"$eq": ["$$fid", "$folderId"]}}}, + {"$project": {"_id": 1}}, + ], + "as": "__items", + } + }, + { + "$lookup": { + "from": "file", + "localField": "__items._id", + "foreignField": "itemId", + "as": "__files", + } + }, + {"$unwind": "$__files"}, + {"$replaceRoot": {"newRoot": "$__files"}}, + ] + ) loadable = hasLoadableFile(files, user=self.getCurrentUser()) return {"loadable": loadable} -def uploadSession(model, parentId, user, size, metadata=None): - # modified from girder.api.v1.file.File.initUpload - parentType = model.__name__.lower() - name = f"session{SESSION_ZIP_EXTENSION}" - try: - # Metadata comes from the client; don't fail the save if the shape - # isn't what we expect. - linkedFilter = (metadata or {}).get("linkedResources", {}).get("filter") - name = sessionNameFromFilter(linkedFilter, SESSION_ZIP_EXTENSION) - except Exception: - pass - - mimeType = "application/zip" - reference = None - parent = model().load(id=parentId, user=user, level=AccessType.WRITE, exc=True) - - chunk = None - ct = cherrypy.request.body.content_type.value - if ( - ct not in cherrypy.request.body.processors - and ct.split("/", 1)[0] not in cherrypy.request.body.processors - ): - chunk = RequestBodyStream(cherrypy.request.body) - if chunk is not None and chunk.getSize() <= 0: - chunk = None - - try: - upload = Upload().createUpload( - user=user, - name=name, - parentType=parentType, - parent=parent, - size=size, - mimeType=mimeType, - reference=reference, - ) - except OSError as exc: - if exc.errno == errno.EACCES: - raise GirderException( - "Failed to create upload.", - f"girder.api.v1.{parentType}.volview_save", - ) - raise - if upload["size"] > 0: - if chunk: - return Upload().handleChunk(upload, chunk, filter=True, user=user) - - return upload - else: - return File().filter(Upload().finalizeUpload(upload), user) - - -@access.public(cookie=True, scope=TokenScope.DATA_WRITE) -@boundHandler -@autoDescribeRoute( - Description("Save VolView session in an item") - .param("itemId", "The item ID", paramType="path") - .errorResponse() -) -def saveToItem(self, itemId): - size = int(cherrypy.request.headers.get("Content-Length")) - if size == 0: - raise GirderException( - "Expected non-zero Content-Length header", "girder.api.v1.item.save-volview" - ) - - return uploadSession(Item, itemId, self.getCurrentUser(), size) - - -@access.public(cookie=True, scope=TokenScope.DATA_WRITE) -@boundHandler -@autoDescribeRoute( - Description("Save VolView session in an folder") - .param("folderId", "The folder ID", paramType="path") - .jsonParam( - "metadata", - "A JSON object containing the metadata keys to add to the item.", - ) - .errorResponse() -) -def saveToFolder(self, folderId, metadata): - user = self.getCurrentUser() - size = int(cherrypy.request.headers.get("Content-Length")) - if size == 0: - raise GirderException( - "Expected non-zero Content-Length header", - "girder.api.v1.folder.volview_save", - ) - # TODO: Should we check if the folder already has an item with a session, - # and, if so, upload to that item rather than calling upload on the - # folder? - fileDic = uploadSession(Folder, folderId, user, size, metadata) - # Ensure next downloadResourcesManifest request for will find this - # session.volview.zip as the freshest session that matches the selection set: - # If there are session.volview.zip items in linked items, - # use its metadata.linkedResources as this item's metadata.linkedResources - linkedResources = metadata["linkedResources"] - linkedItems = normalizeLinkedResources(linkedResources)["items"] - selectedItems = loadModels(user, Item, linkedItems) - newestSelectedSession = findNewestSession(selectedItems) - if newestSelectedSession: - # LinkedResources points to volview.zip. Change saved volview.zip linkedResources - # to match selected linkedResources so we find most recent volview.zip next manifest - # request for those linkedResources. - metadata = {"linkedResources": getLinkedResources(newestSelectedSession)} - - item = Item().load(fileDic["itemId"], user=user, level=AccessType.WRITE, exc=True) - Item().setMetadata(item, metadata) - return fileDic - - -# Deprecated, use downloadManifest -@access.public(cookie=True, scope=TokenScope.DATA_READ) -@boundHandler -@autoDescribeRoute( - Description("Download zip of item files that do not end in volview.zip") - .modelParam("itemId", model=Item, level=AccessType.READ) - .produces(["application/zip"]) - .errorResponse("ID was invalid.") - .errorResponse("Read access was denied for the item.", 403) -) -def downloadDatasets(self, item): - setResponseHeader("Content-Type", "application/zip") - setContentDisposition(item["name"] + ".zip") - - def stream(): - zip = ziputil.ZipGenerator(item["name"]) - sansSessions = [ - fileEntry - for fileEntry in Item().fileList(item, subpath=False, data=False) - if isLoadableImage(fileEntry[1], user=self.getCurrentUser()) - ] - toZip = [ - (path, File().download(file, headers=False)) for path, file in sansSessions - ] - for path, file in toZip: - for data in zip.addFile(file, path): - yield data - yield zip.footer() - - return stream - - @access.public(scope=TokenScope.DATA_READ, cookie=True) @boundHandler @autoDescribeRoute( @@ -336,8 +177,16 @@ def downloadProxiableFile(self, file, name): offset = 0 endByte = None - # to get s3_assetstore_adapter to proxy s3, we set headers to False, - # but to have a correct partial response, fill in headers and status code + # to get s3_assetstore_adapter to proxy s3, we set headers to False, but that + # also suppresses Girder's default download headers. Set safe ones explicitly + # so a proxied file always downloads (attachment) with an inert content type + # and can never render inline in a browser. Transparent to the engine's fetch, + # which reads the response body regardless of these headers. + if proxyRequest: + setResponseHeader("Content-Type", "application/octet-stream") + setContentDisposition(file["name"]) + + # to have a correct partial response, fill in headers and status code if proxyRequest and ( offset > 0 or (endByte is not None and endByte < file["size"]) ): @@ -347,7 +196,7 @@ def downloadProxiableFile(self, file, name): endByte = file["size"] # endByte is non-inclusive, so set Content-Range accordingly cherrypy.response.headers["Content-Range"] = ( - f"bytes {offset}-{endByte-1}/{file['size']}" + f"bytes {offset}-{endByte - 1}/{file['size']}" ) cherrypy.response.headers["Content-Length"] = str(endByte - offset) elif proxyRequest: @@ -359,264 +208,12 @@ def downloadProxiableFile(self, file, name): ) -@access.public(cookie=True, scope=TokenScope.DATA_READ) -@boundHandler -@autoDescribeRoute( - Description( - "Download JSON listing item file download URIs that do not end in volview.zip" - ) - .modelParam("itemId", model=Item, level=AccessType.READ) - .produces(["application/json"]) - .errorResponse("ID was invalid.") - .errorResponse("Read access was denied for the item.", 403) -) -def downloadManifest(self, item): - allFiles = list(Item().fileList(item, subpath=False, data=False)) - files = singleVolViewZipOrImageFiles(allFiles, user=self.getCurrentUser()) - return filesToManifest(files, item["folderId"]) - - -@access.public(cookie=True, scope=TokenScope.DATA_READ) -@boundHandler -@autoDescribeRoute( - Description("Download JSON with file download URIs") - .modelParam("folderId", model=Folder, level=AccessType.READ) - .param("folders", "Folder IDs.", required=False) - .param("items", "Item IDs.", required=False) - .jsonParam("filters", "Filter (dict) or filter list (array of dicts) to apply within a folder.", required=False) - .produces(["application/json"]) - .errorResponse("ID was invalid.") - .errorResponse("Read access was denied for the folders or items.", 403) -) -def downloadResourceManifest(self, folder, folders, items, filters): - user = self.getCurrentUser() - folders = idStringToIdList(folders or '') - items = idStringToIdList(items or '') - # filters is either a dict, a list of dicts, or absent. Anything else - # (bare scalar, null parsed to None already short-circuits via `if not - # filters`) gets rejected here rather than 500ing in Mongo. - if filters is not None and not isinstance(filters, (dict, list)): - raise RestException("filters must be a JSON object or array of objects") - files = [] - if not folders and not items: - if not filters: - # All files in folder (unless volview.zip is found as direct child) - filesInFolder = [ - fileEntry - for fileEntry in Folder().fileList(folder, subpath=False, data=False) - ] - files = singleVolViewZipOrImageFiles( - filesInFolder, user=user, includeFilterLinkedSessions=False, - ) - else: - files = getFilteredSessionFile(folder, filters, user) - if files is None: - filesInFolder = getFilteredFiles(folder, filters) - files = [(None, file) for file in filesInFolder] - else: - # else load selected files - selectedItems = loadModels(user, Item, items) - # If any selected items are session.volview.zip, - # find freshest and change selection set to match its linkedResources. - newestSelectedSession = findNewestSession(selectedItems) - if newestSelectedSession: - linkedResources = getLinkedResources(newestSelectedSession) - linkedFilter = linkedResources.get("filter") - if linkedFilter: - files = getFilteredSessionFile(folder, linkedFilter, user) - if files is None: - files = singleVolViewZipOrImageFiles( - Item().fileList(newestSelectedSession, subpath=False, data=False), - user=user, - ) - return filesToManifest(files, folder["_id"]) - folders = linkedResources["folders"] - items = linkedResources["items"] - - # Find session.volview.zips that match selection set - sessionItems = [ - item for item in Folder().childItems(folder) if isSessionItem(item) - ] - matchingSessionItems = [ - session - for session in sessionItems - if matchesSelectionSet(folders, items, session) - ] - latestSession = getNewestDoc(matchingSessionItems) - # compare touched time of session with max touched time of selected items/folders - selectedFolders = loadModels(user, Folder, folders) - latestSelectedDoc = getNewestDoc(selectedFolders + selectedItems) - if ( - latestSession - and latestSelectedDoc - and getTouchedTime(latestSession) >= getTouchedTime(latestSelectedDoc) - ): - # session touched time is newer than selected items/folders so load it - files = singleVolViewZipOrImageFiles( - Item().fileList(latestSession, subpath=False, data=False), user=user, - ) - else: - # Load selected folders and items excluding child session.volview.zip and .volview_config.yaml - files = getFiles(Folder, selectedFolders) + getFiles(Item, selectedItems) - files = [file for file in files if isLoadableImage(file[1], user)] - return filesToManifest(files, folder["_id"]) - - -def _mergeDictionaries(a, b): - """ - Merge two dictionaries recursively. If the second dictionary (or any - sub-dictionary) has a special key, value of '__all__': True, the updated - dictionary only contains values from the second dictionary and excludes - the __all__ key. - - :param a: the first dictionary. Modified. - :param b: the second dictionary that gets added to the first. - :returns: the modified first dictionary. - """ - if b.get("__all__") is True: - a.clear() - for key in b: - if isinstance(a.get(key), dict) and isinstance(b[key], dict): - _mergeDictionaries(a[key], b[key]) - elif key != "__all__" or b[key] is not True: - a[key] = b[key] - return a - - -def adjustConfigForUser(config, user): - """ - Given the current user, adjust the config so that only relevant and - combined values are used. If the root of the config dictionary contains - "access": {"user": , "admin": }, the base values are updated - based on the user's access level. If the root of the config contains - "group": {: , ...}, the base values are updated for - every group the user is a part of. - - The order of update is groups in C-sort alphabetical order followed by - access/user and then access/admin as they apply. - - :param config: a config dictionary. - """ - if not isinstance(config, dict): - return config - if isinstance(config.get("groups"), dict): - groups = config.pop("groups") - if user: - for group in Group().find( - {"_id": {"$in": user["groups"]}}, sort=[("name", SortDir.ASCENDING)] - ): - if isinstance(groups.get(group["name"]), dict): - config = _mergeDictionaries(config, groups[group["name"]]) - if isinstance(config.get("access"), dict): - accessList = config.pop("access") - if user and isinstance(accessList.get("user"), dict): - config = _mergeDictionaries(config, accessList["user"]) - if user and user.get("admin") and isinstance(accessList.get("admin"), dict): - config = _mergeDictionaries(config, accessList["admin"]) - return config - - -# Modified from https://github.com/girder/large_image/blob/aa1dc05665944e87eb9cb8553085221fab16ae92/girder/girder_large_image/__init__.py#L434-L483 -def yamlConfigFile(folder, name, user, addConfig): - """ - Get a resolved named config file based on a folder and user. - - :param folder: a Girder folder model. - :param name: the name of the config file. - :param user: the user that the response if adjusted for. - :returns: either None if no config file, or a yaml record. - """ - last = False - while folder: - item = Item().findOne({"folderId": folder["_id"], "name": name}) - if item: - for file in Item().childFiles(item): - if file["size"] > 10 * 1024**2: - logger.info("Not loading %s -- too large" % file["name"]) - continue - with File().open(file) as fptr: - config = yaml.safe_load(fptr) - if isinstance(config, list) and len(config) == 1: - config = config[0] - # combine and adjust config values based on current user - if ( - isinstance(config, dict) - and "access" in config - or "group" in config - ): - config = adjustConfigForUser(config, user) - if addConfig and isinstance(config, dict): - config = _mergeDictionaries(config, addConfig) - if ( - not isinstance(config, dict) - or config.get("__inherit__") is not True - ): - return config - config.pop("__inherit__") - addConfig = config - if last: - break - if folder["parentCollection"] != "folder": - if folder["name"] != ".config": - folder = Folder().findOne( - { - "parentId": folder["parentId"], - "parentCollection": folder["parentCollection"], - "name": ".config", - } - ) - else: - last = "setting" - if not folder or last == "setting": - folderId = Setting().get(LARGE_IMAGE_CONFIG_FOLDER) - if not folderId: - break - folder = Folder().load(folderId, force=True) - last = True - else: - folder = Folder().load(folder["parentId"], user=user, level=AccessType.READ) - return addConfig - - -@access.public(cookie=True, scope=TokenScope.DATA_READ) -@boundHandler() -@autoDescribeRoute( - Description("Get a VolView config file.") - .notes( - "Wraps large image yaml_config endpoint and inserts more properties. " - "This walks up the chain of parent folders until the file is found. " - "If not found, the .config folder in the parent collection or user is " - "checked.\n\nAny yaml file can be returned. If the top-level is a " - 'dictionary and contains keys "access" or "groups" where those are ' - "dictionaries, the returned value will be modified based on the " - 'current user. The "groups" dictionary contains keys that are group ' - "names and values that update the main dictionary. All groups that " - "the user is a member of are merged in alphabetical order. If a key " - 'and value of "\\__all\\__": True exists, the replacement is total; ' - 'otherwise it is a merge. If the "access" dictionary exists, the ' - '"user" and "admin" subdictionaries are merged if a calling user is ' - "present and if the user is an admin, respectively (both get merged " - "for admins)." - ) - .modelParam("folderId", model=Folder, level=AccessType.READ) - .param("name", "The name of the file.", paramType="path") - .produces(["application/json"]) - .errorResponse() -) -def getFolderConfigFile(self, folder, name): - user = self.getCurrentUser() - baseConfig = copy.deepcopy(BASE_CONFIG) - config = yamlConfigFile(folder, name, user, None) or {} - config = _mergeDictionaries(baseConfig, config) - return config - - class GirderPlugin(plugin.GirderPlugin): DISPLAY_NAME = "VolView" CLIENT_SOURCE_PATH = "web_client" def load(self, info): - plugin.getPlugin('large_image').load(info) + plugin.getPlugin("large_image").load(info) setupEventHandlers() info["apiRoot"].item.route( @@ -629,15 +226,16 @@ def load(self, info): info["apiRoot"].folder.route( "GET", (":folderId", "volview"), downloadResourceManifest ) + # Session-zip save: item-scoped stuffs the zip into the item, + # folder-scoped creates a new session.volview.zip item. Each returns a + # resumeUrl the client repoints its urls= at, so a later F5 reloads the + # just-made save. + info["apiRoot"].item.route("POST", (":itemId", "volview"), saveToItem) + info["apiRoot"].folder.route("POST", (":folderId", "volview"), saveToFolder) info["apiRoot"].file.route( "GET", (":id", "proxiable", ":name"), downloadProxiableFile ) info["apiRoot"].folder.route( "GET", (":folderId", "volview_config", ":name"), getFolderConfigFile ) - info["apiRoot"].folder.route("POST", (":folderId", "volview"), saveToFolder) - info["apiRoot"].item.route("POST", (":itemId", "volview"), saveToItem) - # volview/datasets is deprecated. Use GET {folder|item}/volview instead. - info["apiRoot"].item.route( - "GET", (":itemId", "volview", "datasets"), downloadDatasets - ) + addBackendRoutes(info) diff --git a/girder_volview/backend/__init__.py b/girder_volview/backend/__init__.py new file mode 100644 index 0000000..175fb8b --- /dev/null +++ b/girder_volview/backend/__init__.py @@ -0,0 +1,3 @@ +from .routes import addBackendRoutes + +__all__ = ["addBackendRoutes"] diff --git a/girder_volview/backend/config.py b/girder_volview/backend/config.py new file mode 100644 index 0000000..845e372 --- /dev/null +++ b/girder_volview/backend/config.py @@ -0,0 +1,57 @@ +"""Processing backend -- the per-launch provider-config block. + +The launch manifest injects this block so the client knows where to reach the +processing provider (``baseUrl`` / ``jobsBaseUrl``) for the folder it opened. +Both URLs derive from the runtime ``getApiRoot()`` so a non-default API mount +still resolves. +""" + +from girder.utility.server import getApiRoot + +# The mount segment of every processing route: the folder-tree routes +# (``/folder/:id/volview_processing/...``) and the folder-free job resource +# (``/volview_processing/...``, see routes.py ``_JobResource``). Shared with the +# route registration so the advertised URLs and the mounted routes cannot drift. +PROCESSING_ROUTE_NAME = "volview_processing" +PROCESSING_PROVIDER_ID_PREFIX = "girder-slicer-cli" + + +def processingProviderId(folderId): + """Return the stable provider identity advertised for a launch folder.""" + return "%s:%s" % (PROCESSING_PROVIDER_ID_PREFIX, folderId) + + +def buildProcessingConfigBlock(folder): + # The block advertises only where to reach the provider, never what is + # loaded: the client mints its own input refs from the on-screen volume's + # provenance. The client zod schema (`src/processing/config.ts` + # processingProviderConfig) reads only id/label/baseUrl/jobsBaseUrl/context + # and strips unknown keys, so an added field stays compatible. + # + # The provider ID is FOLDER-SCOPED and immutable: it carries the launch + # folder id so two folders open simultaneously register as two distinct + # providers (the client keys every job by (providerId, jobId)); a bare + # "girder-slicer-cli" would make both folders share one mutable identity. The + # label carries the folder name so the picker distinguishes them (fall back to + # bare "Analysis" when a folder document has no name). + # + # Both URLs are origin-relative and keyed off getApiRoot() -- the SAME mount + # utils.makeFileDownloadUrl and inputs._fileIdFromMintedUri use. Hardcoding + # "/api/v1" 404s every submit/status/results/stage call on a non-default API + # mount (e.g. /girder/api/v1). jobsBaseUrl is the folder-free root for the + # job-addressed routes (status/results/cancel), advertised explicitly so the + # client never string-surgeries the folder segment out of baseUrl. + folderName = folder.get("name") + return { + "providers": [ + { + "id": processingProviderId(folder["_id"]), + "label": "Analysis — %s" % folderName if folderName else "Analysis", + "baseUrl": ( + f"/{getApiRoot()}/folder/{folder['_id']}/{PROCESSING_ROUTE_NAME}" + ), + "jobsBaseUrl": f"/{getApiRoot()}/{PROCESSING_ROUTE_NAME}", + "context": {}, + } + ] + } diff --git a/girder_volview/backend/inputs.py b/girder_volview/backend/inputs.py new file mode 100644 index 0000000..1b54ffd --- /dev/null +++ b/girder_volview/backend/inputs.py @@ -0,0 +1,399 @@ +"""Processing backend -- input resolution + transient staging lifecycle.""" + +import datetime + +from bson.objectid import ObjectId +from girder import logger +from girder.constants import AccessType +from girder.exceptions import AccessException, RestException +from girder.models.file import File +from girder.models.item import Item +from girder.models.upload import Upload + +# Module-object import (not ``from ... import Job``): call sites resolve +# ``girder_job.Job`` at call time, so tests may monkeypatch the class on +# ``girder_jobs.models.job`` and be seen here. +from girder_jobs.models import job as girder_job + +from ..handles import parseFileHandle +from ..utils import TRANSIENT_STAGED_META_KEY, isTransientStagedItem + +# Every submitted uri is a backend-minted, origin-relative +# ``//file//proxiable/`` (``utils.makeFileDownloadUrl``). +# Resolution recovers the file id from that exact shape and nothing else, then +# re-checks READ access under the submitting user. Type-agnostic: every input +# resolves through this one path — the backend never branches on ``type``. + + +def _fileIdFromMintedUri(uri): + """Recover the Girder file id from a backend-minted proxiable uri, or ``None``. + + Delegates to :func:`girder_volview.handles.parseFileHandle`, the ONE parse + site for the load-handle scheme and the exact mirror of the mint + (``handles.mintFileHandle`` / ``utils.makeFileDownloadUrl``). Returns + ``None`` for anything outside the backend's own scheme, so a foreign or + malformed string is rejected by the caller and never dereferenced. + """ + parsed = parseFileHandle(uri) + return parsed[0] if parsed else None + + +def resolveInputUrisToFiles(uris, user): + """Resolve a client-minted uri list to readable Girder files (fail closed). + + A uri that does not match the backend's mint is rejected 400 and never + fetched, and every recovered id is loaded with the submitting user's READ + permission, so possession of a (by-design recoverable) id is not itself a + capability. Validation runs over every uri first, then authorization, so a + malformed uri fails 400 ahead of an unreadable id's 403. + """ + if not isinstance(uris, list) or not uris: + raise RestException("Processing input value carries no uris", code=400) + fileIds = [] + for uri in uris: + fileId = _fileIdFromMintedUri(uri) + if fileId is None: + raise RestException( + "Processing input uri does not match this server's file scheme", + code=400, + ) + fileIds.append(fileId) + return _readableFilesInOrder(fileIds, user) + + +def readableFilesById(fileObjectIds, user, fields=None): + """Load the file docs whose parent item ``user`` can READ, batched. + + THE one ACL boundary for bulk file reads: a file is readable iff its parent + item is READable. Girder files inherit access through their parent item, and + ``File().load`` with a user+level runs a per-file ACL that falls back to + loading that parent -- ~2 Mongo queries EACH (≈600 for a 300-slice DICOM + series). Batched instead: one ``File().find`` for every id, then ONE + permission-filtered ``Item().findWithPermissions`` over the distinct parent + items. Returns ``{str(fileId): fileDoc}``; a missing file, missing parent, + or unreadable parent is simply absent, and lenient/strict handling stays + with the callers. + """ + fileDocs = list( + File().find(query={"_id": {"$in": list(fileObjectIds)}}, fields=fields) + ) + itemIds = {fileDoc.get("itemId") for fileDoc in fileDocs} + itemIds.discard(None) + if not itemIds: + return {} + readableItemIds = { + itemDoc["_id"] + for itemDoc in Item().findWithPermissions( + query={"_id": {"$in": list(itemIds)}}, + fields={"_id": 1}, + user=user, + level=AccessType.READ, + ) + } + return { + str(fileDoc["_id"]): fileDoc + for fileDoc in fileDocs + if fileDoc.get("itemId") in readableItemIds + } + + +def _readableFilesInOrder(fileIds, user): + """Load READ-authorized file docs for ``fileIds``, in input order, raising. + + A strict adapter over :func:`readableFilesById`: any unreadable id raises + ``AccessException``. ``_fileIdFromMintedUri`` already validated each id's + shape, so the ``ObjectId`` conversion cannot fail here. Ordering matches the + input ``fileIds`` (a comma-joined multi-file volume forwards ids + positionally), and a repeated id resolves to the same doc. + """ + filesById = readableFilesById([ObjectId(fileId) for fileId in fileIds], user) + files = [] + for fileId in fileIds: + fileDoc = filesById.get(fileId) + if fileDoc is None: + # Missing file, missing parent, or a parent the user cannot READ: + # possession of a (by-design recoverable) id is not a capability. + raise AccessException("Read access denied for file %s." % fileId) + files.append(fileDoc) + return files + + +def validateStagedDescriptor(descriptor, user): + """Validate a staged-resource descriptor end to end; return its name. + + Owns the whole descriptor schema — shape, the ``labelmap`` type + discriminator, and the reference image (own-scheme + ACL + durable) — so the + ``stageInput`` route stays transport + authorization only and a future + staged type extends this one validator. + """ + if set(descriptor) != {"type", "name", "referenceImage"}: + raise RestException("Malformed staged resource descriptor", code=400) + if descriptor.get("type") != "labelmap": + raise RestException("Staged resource type must be labelmap", code=400) + name = descriptor.get("name") + if not isinstance(name, str) or not name: + raise RestException("Staged resource name must not be empty", code=400) + referenceImage = descriptor.get("referenceImage") + if not isinstance(referenceImage, dict): + raise RestException("Staged labelmap requires a reference image", code=400) + if not set(referenceImage).issubset({"type", "format", "uris"}): + raise RestException("Malformed staged reference image", code=400) + if "format" in referenceImage and not isinstance(referenceImage["format"], str): + raise RestException("Malformed staged reference image format", code=400) + validateStagedReferenceImage(referenceImage, user) + return name + + +def validateStagedReferenceImage(referenceImage, user): + """Validate a staged labelmap's reference image (own-scheme + ACL + durable). + + Resolves the reference image's own-scheme uris to files under the caller's + READ permission — rejecting a malformed, foreign, or unauthorized reference + — and rejects a transient reference so a staged labelmap never binds to + ephemeral data. Validation only: no lineage is tracked. + """ + if not isinstance(referenceImage, dict) or referenceImage.get("type") != "image": + raise RestException("Staged labelmap requires a reference image", code=400) + fileDocs = resolveInputUrisToFiles(referenceImage.get("uris"), user) + itemIds = {fileDoc.get("itemId") for fileDoc in fileDocs} + itemIds.discard(None) + # ``resolveInputUrisToFiles`` already enforced READ on each file's parent item, + # so this read only inspects the transient marker: one batched find over the + # DISTINCT parent items replaces a per-file (and per-duplicate) ``Item().load``. + for item in Item().find(query={"_id": {"$in": list(itemIds)}}): + if _isTransientItem(item): + raise RestException( + "Staged labelmap requires a durable reference image", code=400 + ) + + +# ``stageInput`` (in ``routes.py``) lands client-held bytes in a fresh item +# tagged transient and mints a proxiable download URI for them; from there a +# staged input resolves through the same own-scheme path as any other input. +# Ownership is per-job by construction (``copyStagedInputsIntoJobFolder``), so +# no job references a shared staged original. Cleanup is therefore split: the +# job deletes its own copies at terminal state, and the TTL sweep below ages +# out the originals, which have no job to clean them up. + +# Age after which an uploaded-but-never-submitted transient item is swept on the +# next staging call. Upload->submit is normally seconds; a day absorbs an +# interrupted session without cluttering folders across days. +_TRANSIENT_ORPHAN_TTL = datetime.timedelta(hours=24) + + +# One definition of the staging predicate, shared with the launch-manifest +# exclusion (``utils.isTransientStagedFile``); aliased so this module's call +# sites and test doubles keep their name. +_isTransientItem = isTransientStagedItem + + +def copyStagedInputsIntoJobFolder(params, resolvedInputFiles, user, outputFolder): + """Give the job its OWN copies of any staged (transient) inputs. + + Every transient staged item among a submission's bound inputs is copied + into the job's private folder and the CLI file-id params are rewritten + onto the copies, so the job references only resources it alone owns and two + jobs reusing one staged original can never delete each other's inputs. The + original stays covered by the TTL orphan sweep. Transience is decided by + the parent item's marker, never by ``type``. + + URI resolution (``resolveInputUrisToFiles``) already enforced READ on every + parent item moments ago, so the transient markers are read with ONE batched + find over the distinct parents rather than a per-item ACL'd load. A parent + that vanished in between (an orphan sweep or delete) raises 409 rather than + publishing a job against deleted file ids. ``Item().copyItem`` deep-copies + metadata, so a copy carries the transient marker and is cleaned up exactly + like any staged item. + + Returns ``(params, copiedItemIds)`` — the (possibly rewritten) params and + the copied item ids to record on the job for terminal cleanup. + """ + itemIds = { + fileDoc["itemId"] + for fileDocs in resolvedInputFiles.values() + for fileDoc in fileDocs + if (fileDoc or {}).get("itemId") + } + items = list(Item().find({"_id": {"$in": list(itemIds)}})) if itemIds else [] + if len(items) != len(itemIds): + # URI resolution ACL-loaded these parents moments ago, so a missing item + # means a concurrent delete won the race. Fail the submit rather than + # publish a job whose params reference deleted files. + raise RestException( + "A processing input was removed while the submission " + "was in progress; please resubmit", + code=409, + ) + fileIdRemap = {} + copiedItemIds = [] + for item in items: + if not _isTransientItem(item): + continue + copied = Item().copyItem(item, creator=user, folder=outputFolder) + copiedItemIds.append(str(copied["_id"])) + # Copied files preserve name/size/checksum; sorting both sides by that + # triple pairs each original with its copy regardless of the underlying + # cursor order. Girder permits same-named files in one item, so name + # alone could pair A with B's copy — with the full triple, files that + # still tie are byte-identical and any pairing is correct. copyItem + # duplicates every child file, so a length mismatch means the staged + # item's files changed between resolution and copy — the same race as + # the concurrent-delete guard above, and the same typed 409 rather than + # running the job against a partial input. + def pairingKey(f): + return (f.get("name", ""), f.get("size", 0), f.get("sha512") or "") + + originals = sorted(Item().childFiles(item), key=pairingKey) + copies = sorted(Item().childFiles(copied), key=pairingKey) + if len(originals) != len(copies): + raise RestException( + "A processing input changed while the submission " + "was in progress; please resubmit", + code=409, + ) + fileIdRemap.update( + { + str(orig["_id"]): str(cop["_id"]) + for orig, cop in zip(originals, copies, strict=True) + } + ) + if not fileIdRemap: + return params, copiedItemIds + params = dict(params) + for paramName, fileDocs in resolvedInputFiles.items(): + params[paramName] = ",".join( + fileIdRemap.get(str(fileDoc["_id"]), str(fileDoc["_id"])) + for fileDoc in fileDocs + ) + return params, copiedItemIds + + +# Girder jobs are user-owned, not folder-linked, so `listJobHistory` can only +# scope to a launch folder if the context is stamped ON the job at submit, as +# plain otherFields (queryable Mongo keys). The task id is backend association +# data the history summary does not expose. +_LAUNCH_FOLDER_FIELD = "volviewLaunchFolderId" # str(folder _id) — scope key +_TASK_ID_FIELD = "volviewTaskId" + + +def _removeTransientItems(itemIds): + """Delete transient input items by id (idempotent, best-effort).""" + for itemId in itemIds: + try: + item = Item().load(itemId, force=True) + if item: + Item().remove(item) + except Exception: + logger.exception("Failed to remove transient item %s", itemId) + + +def _cleanupTransientOnJobDone(event): + """Delete a job's transient staged inputs once it reaches a terminal state. + + Bound to ``jobs.job.update.after``. Idempotent: a re-fired terminal update + finds the items already gone and no-ops. A present, non-terminal in-memory + status short-circuits before any DB work, so the common progress/log tick + costs nothing. When the in-memory status is terminal or absent, the job is + reloaded from the database before reading the marker/status -- the event + carries the updater's *in-memory* job dict, which holds the marker only if + that updater happened to DB-load the job first. Reloading keeps cleanup + self-contained for any terminal updater (girder_worker, a manual cancel). + """ + + from .results import isTerminalStatus + + info = getattr(event, "info", None) + eventJob = info.get("job") if isinstance(info, dict) else None + if not isinstance(eventJob, dict): + return + # This handler fires on EVERY job update instance-wide (progress ticks, log + # appends). ``updateJob`` sets the new status ON the in-memory job dict + # before firing, so a present, non-terminal in-memory status is an + # authoritative "not settled yet" -- short-circuit before the DB reload the + # steady-state stream would otherwise pay on every tick. + inMemoryStatus = eventJob.get("status") + if inMemoryStatus is not None and not isTerminalStatus(inMemoryStatus): + return + # Reload the committed doc to read the marker/status self-containedly -- the + # event's in-memory job dict may carry neither. includeLog=False because + # only the marker + status are read; loading the unbounded log would + # re-materialize it out of Mongo on every tick. + job = girder_job.Job().load(eventJob.get("_id"), force=True, includeLog=False) + if not isinstance(job, dict): + return + transientItemIds = job.get(TRANSIENT_STAGED_META_KEY) + if not isinstance(transientItemIds, list) or not transientItemIds: + return + if not isTerminalStatus(job.get("status")): + return + _removeTransientItems(transientItemIds) + + +def _sweepOrphanTransients(folder, now=None): + """Age out stale transient items in ``folder`` (best-effort). + + Piggybacked on staging calls (an upload precedes its job, so job-end cleanup + never sees a never-submitted orphan). Keyed off ``item['created']`` because + the marker carries no timestamp; only items strictly older than + :data:`_TRANSIENT_ORPHAN_TTL` are candidates, so the item this same call is + about to create is never one. Age alone decides: no job ever depends on a + staged ORIGINAL — submission rewires the job onto its own private copies + (:func:`copyStagedInputsIntoJobFolder`), which live in the job's private + folder, not the staging folder this sweep scans. + """ + now = now or datetime.datetime.utcnow() + cutoff = now - _TRANSIENT_ORPHAN_TTL + query = { + "folderId": folder["_id"], + "meta.%s" % TRANSIENT_STAGED_META_KEY: True, + "created": {"$lt": cutoff}, + } + try: + stale = list(Item().find(query)) + except Exception: + logger.exception("Failed to query orphan transient items") + return + for item in stale: + try: + Item().remove(item) + except Exception: + logger.exception( + "Failed to sweep orphan transient item %s", item.get("_id") + ) + + +def _streamMultipartFileIntoItem(folder, user, part, name): + """Stream one parsed multipart file part into a fresh item under ``folder``. + + ``folder`` is the WRITE-authorized document the ``stageInput`` route already + loaded via its ``modelParam(level=AccessType.WRITE)`` decorator — the single + authorization boundary, deliberately not re-checked here. + """ + stream = getattr(part, "file", None) + if stream is None: + raise RestException("Staging request carries no file part", code=400) + stream.seek(0, 2) + size = stream.tell() + stream.seek(0) + if size <= 0: + raise RestException("Staging file must not be empty", code=400) + return Upload().uploadFromFile( + stream, + size=size, + name=name, + parentType="folder", + parent=folder, + user=user, + mimeType="application/octet-stream", + ) + + +def _tagItemTransient(fileDoc): + """Tag a freshly-uploaded file's parent item transient; return the item.""" + itemId = fileDoc.get("itemId") + if not itemId: + return None + item = Item().load(itemId, force=True) + if item: + Item().setMetadata(item, {TRANSIENT_STAGED_META_KEY: True}) + return item diff --git a/girder_volview/backend/launch.py b/girder_volview/backend/launch.py new file mode 100644 index 0000000..12a0683 --- /dev/null +++ b/girder_volview/backend/launch.py @@ -0,0 +1,510 @@ +"""Launch / compose / config / save handlers for the ordinary VolView viewer. + +Each launch gesture has one meaning: raw checked picks ALWAYS open fresh; a +checked session item opens through to exactly that session; a filter gesture +resumes its newest matching session; a bare folder-open resumes the folder's +newest unfiltered session. A save returns a ``resumeUrl`` the client uses for +subsequent reloads (F5); the save target itself stays launch-provided, so +folder saves mint a new ``session.volview.zip`` item per save. +""" + +import copy +import errno + +import cherrypy +import yaml + +from girder import logger +from girder.api import access +from girder.api.describe import Description, autoDescribeRoute +from girder.api.rest import boundHandler +from girder.constants import AccessType, TokenScope, SortDir +from girder.exceptions import GirderException, RestException +from girder.models.file import File +from girder.models.folder import Folder +from girder.models.group import Group +from girder.models.item import Item +from girder.models.setting import Setting +from girder.models.upload import Upload +from girder.utility import RequestBodyStream +from girder.utility.server import getApiRoot + +from .config import buildProcessingConfigBlock +from ..utils import ( + SESSION_ZIP_EXTENSION, + isJobOutputFolderItem, + isLaunchFile, + isLoadableImage, + primeLoadableImageCaches, + filesToManifest, + singleVolViewZipOrImageFiles, + getFilteredFiles, + getFilteredSessionFile, + getFiles, + getLinkedResources, + idStringToIdList, + findNewestSession, + loadModels, + normalizeLinkedResources, + sessionNameFromFilter, +) + +LARGE_IMAGE_CONFIG_FOLDER = "large_image.config_folder" + +BASE_CONFIG = { + "io": { + "segmentGroupExtension": "seg", + "segmentGroupSaveFormat": "nii.gz", + "layerExtension": "layer", + }, + "disabledViewTypes": ["3D", "Oblique"], + "layouts": { + "Axial Coronal Sagittal": { + "direction": "row", + "items": [ + "axial", + {"direction": "column", "items": ["coronal", "sagittal"]}, + ], + }, + "Axial Only": [["axial"]], + }, +} + + +def uploadSession(model, parentId, user, size, metadata=None): + # modified from girder.api.v1.file.File.initUpload + parentType = model.__name__.lower() + name = f"session{SESSION_ZIP_EXTENSION}" + try: + # Metadata comes from the client; don't fail the save if the shape + # isn't what we expect. + linkedFilter = (metadata or {}).get("linkedResources", {}).get("filter") + name = sessionNameFromFilter(linkedFilter, SESSION_ZIP_EXTENSION) + except Exception: + pass + + mimeType = "application/zip" + reference = None + parent = model().load(id=parentId, user=user, level=AccessType.WRITE, exc=True) + + chunk = None + ct = cherrypy.request.body.content_type.value + if ( + ct not in cherrypy.request.body.processors + and ct.split("/", 1)[0] not in cherrypy.request.body.processors + ): + chunk = RequestBodyStream(cherrypy.request.body) + if chunk is not None and chunk.getSize() <= 0: + chunk = None + + try: + upload = Upload().createUpload( + user=user, + name=name, + parentType=parentType, + parent=parent, + size=size, + mimeType=mimeType, + reference=reference, + ) + except OSError as exc: + if exc.errno == errno.EACCES: + raise GirderException( + "Failed to create upload.", + f"girder.api.v1.{parentType}.volview_save", + ) from exc + raise + if upload["size"] > 0: + if chunk: + return Upload().handleChunk(upload, chunk, filter=True, user=user) + + return upload + else: + return File().filter(Upload().finalizeUpload(upload), user) + + +def _saveResponse(sessionItemId): + """The save response — a SINGLE field: the session's load URL. + + The VolView client stays opaque to Girder ids: it never learns the item id, + only the ``resumeUrl`` (``item/:id/volview``), which it repoints ONLY its + reload (``urls=``) at. So a later F5 reloads exactly this save, while the + save target (``save=``) stays launch-provided — a folder-scoped save mints + a new ``session.volview.zip`` item on every save. + """ + return {"resumeUrl": f"/{getApiRoot()}/item/{sessionItemId}/volview"} + + +def _uploadWholeSession(model, parentId, user, errorIdentifier, metadata=None): + """Upload the session zip in one shot; 400 unless it finalized into a File. + + Only a finalized File carries ``itemId``. A resumable/partial upload (a + processor content-type, or a body shorter than the declared Content-Length) + returns the raw Upload doc with no ``itemId``; the single-shot save contract + can't continue, so fail with a clean 400 rather than report a success F5 + would contradict by restoring the previous zip (or a KeyError -> 500 and an + orphaned item). + """ + try: + size = int(cherrypy.request.headers.get("Content-Length")) + except (TypeError, ValueError): + # Absent (e.g. Transfer-Encoding: chunked) or non-integer header: + # the same clean rejection as an empty body, not an int() 500. + size = 0 + if size == 0: + raise GirderException( + "Expected non-zero Content-Length header", errorIdentifier + ) + fileDic = uploadSession(model, parentId, user, size, metadata) + if "itemId" not in fileDic: + raise RestException( + "Session save must upload the whole zip in one request.", code=400 + ) + return fileDic + + +@access.public(cookie=True, scope=TokenScope.DATA_WRITE) +@boundHandler +@autoDescribeRoute( + Description("Save VolView session in an item") + .param("itemId", "The item ID", paramType="path") + .errorResponse() +) +def saveToItem(self, itemId): + _uploadWholeSession( + Item, itemId, self.getCurrentUser(), "girder.api.v1.item.save-volview" + ) + # The session file is stuffed into this same item, so its own manifest URL + # is both the save target and the F5 reload target. + return _saveResponse(itemId) + + +@access.public(cookie=True, scope=TokenScope.DATA_WRITE) +@boundHandler +@autoDescribeRoute( + Description("Save VolView session in an folder") + .param("folderId", "The folder ID", paramType="path") + .jsonParam( + "metadata", + "A JSON object containing the metadata keys to add to the item.", + ) + .errorResponse() +) +def saveToFolder(self, folderId, metadata): + user = self.getCurrentUser() + # jsonParam yields whatever the client sent, so `metadata` can be a list or + # scalar. Coerce before the upload -- same "don't fail the save on an + # unexpected shape" stance as uploadSession, and a guard placed after the + # upload would raise only once the zip is stored, 500ing on an orphan item. + if not isinstance(metadata, dict): + metadata = {} + # Rebase this save's linkedResources onto the newest already-saved session in + # the selection set: a save from a checked-session open would otherwise stamp + # linkedResources={items:[S]} instead of S's own lineage. Load-bearing for + # filter sessions — a save from a checked FILTER-session open must inherit + # the filter link so the filter row resumes this newest save and the bare + # folder-open keeps excluding it. + # + # Resolved BEFORE the upload: a malformed linkedResources (truthy non-object) + # or an unloadable id must 4xx with nothing stored, not raise once the zip has + # already finalized into a session item that a folder-open would then pick as + # the newest session to restore. + rawLinked = metadata.get("linkedResources") + linkedResources = normalizeLinkedResources( + rawLinked if isinstance(rawLinked, dict) else None + ) + selectedItems = loadModels(user, Item, linkedResources["items"]) + newestSelectedSession = findNewestSession(selectedItems) + savedMetadata = metadata + if newestSelectedSession: + savedMetadata = {"linkedResources": getLinkedResources(newestSelectedSession)} + + fileDic = _uploadWholeSession( + Folder, folderId, user, "girder.api.v1.folder.volview_save", metadata + ) + item = Item().load(fileDic["itemId"], user=user, level=AccessType.WRITE, exc=True) + try: + Item().setMetadata(item, savedMetadata) + except Exception: + # An unstamped session item is still the folder's newest session, so a + # later folder-open would restore this failed save. Drop it instead. + Item().remove(item) + raise + return _saveResponse(fileDic["itemId"]) + + +@access.public(cookie=True, scope=TokenScope.DATA_READ) +@boundHandler +@autoDescribeRoute( + Description( + "Download the VolView launch manifest for an item: a session.volview.zip " + "item opens through (restore); any other item resolves to its loadable " + "images (fresh)." + ) + .modelParam("itemId", model=Item, level=AccessType.READ) + .produces(["application/json"]) + .errorResponse("ID was invalid.") + .errorResponse("Read access was denied for the item.", 403) +) +def downloadManifest(self, item): + user = self.getCurrentUser() + # Job outputs stay durable in the folder but out of the launch manifest: a + # direct open of an item inside a job's private output folder yields nothing. + if isJobOutputFolderItem(item): + return filesToManifest([], item["folderId"]) + allFiles = list(Item().fileList(item, subpath=False, data=False)) + # A session file opens through (restore); otherwise the item's raw images. + files = singleVolViewZipOrImageFiles( + allFiles, user=user, itemCache={item["_id"]: item}, folderCache={} + ) + return filesToManifest(files, item["folderId"]) + + +@access.public(cookie=True, scope=TokenScope.DATA_READ) +@boundHandler +@autoDescribeRoute( + Description( + "Download the VolView launch manifest for a folder / checked / filter " + "gesture: a checked session item opens through to exactly that session " + "(back-in-history); raw checked items/folders ALWAYS open fresh; a " + "filter gesture resumes its newest matching session when one exists; a " + "bare folder-open resumes the folder's newest session.volview.zip, " + "else all its raw images. An explicit folders/items " + "selection takes precedence over filters; filters apply only when no " + "selection is passed, and the filtered leg returns only loadable images " + "(transient staged inputs and job-output-folder files are excluded)." + ) + .modelParam("folderId", model=Folder, level=AccessType.READ) + .param("folders", "Folder IDs.", required=False) + .param("items", "Item IDs.", required=False) + .jsonParam( + "filters", + "Filter (dict) or filter list (array of dicts) to apply within a folder.", + required=False, + ) + .produces(["application/json"]) + .errorResponse("ID was invalid.") + .errorResponse("Read access was denied for the folders or items.", 403) +) +def downloadResourceManifest(self, folder, folders, items, filters): + user = self.getCurrentUser() + itemCache = {} + folderCache = {} + folders = idStringToIdList(folders or "") + items = idStringToIdList(items or "") + # filters is either a dict, a list of dicts, or absent. Anything else + # (bare scalar) is rejected here rather than 500ing in Mongo. + if filters is not None and not isinstance(filters, (dict, list)): + raise RestException("filters must be a JSON object or array of objects") + # An explicit folders/items selection wins over filters: a stale/bookmarked + # URL carrying both must load the checked resources, not silently + # substitute the filter set. + if folders or items: + selectedItems = loadModels(user, Item, items) + checkedSession = findNewestSession(selectedItems) + if checkedSession: + # An explicitly checked session item opens through to EXACTLY that + # session — the back-in-history gesture. Never re-match it to a + # newer sibling save. Filter-linked sessions get the same treatment: + # re-entering the filter row resumes the newest, but checking an old + # one opens exactly it. + files = singleVolViewZipOrImageFiles( + Item().fileList(checkedSession, subpath=False, data=False), + user=user, + itemCache=itemCache, + folderCache=folderCache, + ) + return filesToManifest(files, folder["_id"]) + + # Raw checked picks ALWAYS open fresh: checking images is the "start + # fresh" gesture, so no saved session is ever substituted. Resume + # happens only through the other gestures — bare folder-open (newest + # save), checking a session item (exactly that save), or re-entering a + # filter row (newest filter save). + selectedFolders = loadModels(user, Folder, folders) + files = getFiles(Folder, selectedFolders) + getFiles(Item, selectedItems) + primeLoadableImageCaches([f[1] for f in files], user, itemCache, folderCache) + files = [ + f for f in files if isLoadableImage(f[1], user, itemCache, folderCache) + ] + elif filters: + files = getFilteredSessionFile(folder, filters, user) + # Empty is treated as no-match, not as a resolved session: a matched + # session whose files are all unloadable would otherwise skip the fresh + # leg and emit a manifest of nothing but config.json — a blank viewer + # with no gesture that recovers it. + if not files: + files = getFilteredFiles(folder, filters) + # The filter row owns every file it matched — no loadability gate + # (grouped DICOM rows carry extensionless slices). Only working + # data is excluded: transient staged inputs and session zips. + primeLoadableImageCaches(files, user, itemCache, folderCache) + files = [ + (None, f) + for f in files + if isLaunchFile(f, user, itemCache, folderCache) + ] + else: + # Bare folder-open -> resume the folder's newest session.volview.zip, + # else all its raw images. Filter-linked sessions are excluded (they are + # only meaningful re-entered through their filter). + filesInFolder = list(Folder().fileList(folder, subpath=False, data=False)) + files = singleVolViewZipOrImageFiles( + filesInFolder, + user=user, + includeFilterLinkedSessions=False, + itemCache=itemCache, + folderCache=folderCache, + ) + return filesToManifest(files, folder["_id"]) + + +def _mergeDictionaries(a, b): + """ + Merge two dictionaries recursively. If the second dictionary (or any + sub-dictionary) has a special key, value of '__all__': True, the updated + dictionary only contains values from the second dictionary and excludes + the __all__ key. + + :param a: the first dictionary. Modified. + :param b: the second dictionary that gets added to the first. + :returns: the modified first dictionary. + """ + if b.get("__all__") is True: + a.clear() + for key in b: + if isinstance(a.get(key), dict) and isinstance(b[key], dict): + _mergeDictionaries(a[key], b[key]) + elif key != "__all__" or b[key] is not True: + a[key] = b[key] + return a + + +def adjustConfigForUser(config, user): + """ + Given the current user, adjust the config so that only relevant and + combined values are used. If the root of the config dictionary contains + "access": {"user": , "admin": }, the base values are updated + based on the user's access level. If the root of the config contains + "group": {: , ...}, the base values are updated for + every group the user is a part of. + + The order of update is groups in C-sort alphabetical order followed by + access/user and then access/admin as they apply. + + :param config: a config dictionary. + """ + if not isinstance(config, dict): + return config + if isinstance(config.get("groups"), dict): + groups = config.pop("groups") + if user: + for group in Group().find( + {"_id": {"$in": user["groups"]}}, sort=[("name", SortDir.ASCENDING)] + ): + if isinstance(groups.get(group["name"]), dict): + config = _mergeDictionaries(config, groups[group["name"]]) + if isinstance(config.get("access"), dict): + accessList = config.pop("access") + if user and isinstance(accessList.get("user"), dict): + config = _mergeDictionaries(config, accessList["user"]) + if user and user.get("admin") and isinstance(accessList.get("admin"), dict): + config = _mergeDictionaries(config, accessList["admin"]) + return config + + +# Modified from https://github.com/girder/large_image/blob/aa1dc05665944e87eb9cb8553085221fab16ae92/girder/girder_large_image/__init__.py#L434-L483 +def yamlConfigFile(folder, name, user, addConfig): + """ + Get a resolved named config file based on a folder and user. + + :param folder: a Girder folder model. + :param name: the name of the config file. + :param user: the user that the response if adjusted for. + :returns: either None if no config file, or a yaml record. + """ + last = False + while folder: + item = Item().findOne({"folderId": folder["_id"], "name": name}) + if item: + for file in Item().childFiles(item): + if file["size"] > 10 * 1024**2: + logger.info("Not loading %s -- too large" % file["name"]) + continue + with File().open(file) as fptr: + config = yaml.safe_load(fptr) + if isinstance(config, list) and len(config) == 1: + config = config[0] + # combine and adjust config values based on current user + if isinstance(config, dict) and ( + "access" in config or "groups" in config + ): + config = adjustConfigForUser(config, user) + if addConfig and isinstance(config, dict): + config = _mergeDictionaries(config, addConfig) + if ( + not isinstance(config, dict) + or config.get("__inherit__") is not True + ): + return config + config.pop("__inherit__") + addConfig = config + if last: + break + if folder["parentCollection"] != "folder": + if folder["name"] != ".config": + folder = Folder().findOne( + { + "parentId": folder["parentId"], + "parentCollection": folder["parentCollection"], + "name": ".config", + } + ) + else: + last = "setting" + if not folder or last == "setting": + folderId = Setting().get(LARGE_IMAGE_CONFIG_FOLDER) + if not folderId: + break + folder = Folder().load(folderId, force=True) + last = True + else: + folder = Folder().load(folder["parentId"], user=user, level=AccessType.READ) + return addConfig + + +@access.public(cookie=True, scope=TokenScope.DATA_READ) +@boundHandler() +@autoDescribeRoute( + Description("Get a VolView config file.") + .notes( + "Wraps large image yaml_config endpoint and inserts more properties. " + "This walks up the chain of parent folders until the file is found. " + "If not found, the .config folder in the parent collection or user is " + "checked.\n\nAny yaml file can be returned. If the top-level is a " + 'dictionary and contains keys "access" or "groups" where those are ' + "dictionaries, the returned value will be modified based on the " + 'current user. The "groups" dictionary contains keys that are group ' + "names and values that update the main dictionary. All groups that " + "the user is a member of are merged in alphabetical order. If a key " + 'and value of "\\__all\\__": True exists, the replacement is total; ' + 'otherwise it is a merge. If the "access" dictionary exists, the ' + '"user" and "admin" subdictionaries are merged if a calling user is ' + "present and if the user is an admin, respectively (both get merged " + "for admins)." + ) + .modelParam("folderId", model=Folder, level=AccessType.READ) + .param("name", "The name of the file.", paramType="path") + .produces(["application/json"]) + .errorResponse() +) +def getFolderConfigFile(self, folder, name): + user = self.getCurrentUser() + baseConfig = copy.deepcopy(BASE_CONFIG) + config = yamlConfigFile(folder, name, user, None) or {} + config = _mergeDictionaries(baseConfig, config) + # Injected dynamically rather than living in BASE_CONFIG: the providers list + # depends on the folder being launched. + processing = buildProcessingConfigBlock(folder) + _mergeDictionaries(config, {"processing": processing}) + return config diff --git a/girder_volview/backend/outputs.py b/girder_volview/backend/outputs.py new file mode 100644 index 0000000..18155a0 --- /dev/null +++ b/girder_volview/backend/outputs.py @@ -0,0 +1,401 @@ +"""Processing backend -- folder-owned job-output correlation (write + cleanup). + +Each backend job OWNS exactly one server-created private output folder. Every +output ``girder_worker`` uploads is forced into that folder, so an upload is +correlated back to its job by the finalized file's ACTUAL parent folder -- never +by a filename, a token, or a caller-supplied job id (all of which are +attacker-controllable). The read path (projecting recorded ids into result +intents) and status projection live in ``results.py``. + +Ownership is also the deletion boundary, and it cascades both ways: a +``model.job.remove`` handler deletes the owned output folder plus any remaining +staged inputs and REFUSES a non-terminal owned job, while a +``model.folder.remove`` handler deletes the owning job when its output folder is +removed in the Girder hierarchy. Girder fires these handlers synchronously +BEFORE the DB delete and wraps them in no try/except, so raising aborts the +removal and retains the job record as the discoverable owner of whatever is left. +""" + +import json + +from girder.exceptions import RestException +from girder.models.folder import Folder +from girder.models.item import Item + +# Module-object import (not ``from ... import Job``): call sites resolve +# ``girder_job.Job`` at call time, so tests may monkeypatch the class on +# ``girder_jobs.models.job`` and be seen here. +from girder_jobs.models import job as girder_job + +from ..utils import JOB_OUTPUT_FOLDER_META_KEY, TRANSIENT_STAGED_META_KEY +from .inputs import _removeTransientItems + +# Backend-owned job fields (otherFields, not a schema change). The id map is +# READ-exposed (``routes.addBackendRoutes``); the job's own ACL is the gate. +_OUTPUTS_FIELD = "volviewOutputs" # {identifier: str(fileId)} +_OUTPUT_SPECS_FIELD = "volviewOutputSpecs" # [{name, tag, isLabel, fileExtensions}] +_OUTPUT_FOLDER_ID_FIELD = "volviewOutputFolderId" # str(folder _id) -- the job's +# private output folder; the SOLE output-correlation and ownership key. + + +def _parseOutputReference(raw): + """Decode an upload reference to a dict, or ``None``. + + ``girder_worker`` stamps each output upload with slicer_cli_web's JSON + reference (``prepare_task.py`` -- carries the output ``identifier``). A + non-JSON / non-dict / identifier-less reference (a foreign upload, or the + backend's own reference-less staging upload) yields ``None`` so the handler + skips it (fail closed). An identifier carrying ``.`` or ``$`` is rejected too, + so it can never be used to build an unintended nested / operator ``$set`` key. + """ + if raw is None: + return None + try: + ref = raw if isinstance(raw, dict) else json.loads(raw) + except (ValueError, TypeError): + return None + if not isinstance(ref, dict): + return None + identifier = ref.get("identifier") + if not isinstance(identifier, str) or not identifier: + return None + if "." in identifier or "$" in identifier: + return None + return ref + + +def _declaredOutputIdentifiers(job): + """The set of output identifiers the job DECLARED at submit (or empty). + + An upload whose identifier is not one this job declared is never recorded -- + a crafted reference cannot introduce an undeclared output key. + """ + specs = (job or {}).get(_OUTPUT_SPECS_FIELD) + if not isinstance(specs, list): + return set() + return { + spec["name"] + for spec in specs + if isinstance(spec, dict) and isinstance(spec.get("name"), str) + } + + +def _jobForOutputFolder(folderId): + """Find the job that OWNS a given output folder, or ``None`` (fail closed). + + Correlation is solely by the finalized file's actual parent folder: each job + owns exactly one private output folder, so the parent folder uniquely + identifies the job. A caller-supplied jobId / uuid / token / filename in the + upload reference is NEVER consulted. The id is stored as a string on the job, + so the query stringifies the parent folder id to match. + """ + + if folderId is None: + return None + job = girder_job.Job().findOne({_OUTPUT_FOLDER_ID_FIELD: str(folderId)}) + return job if isinstance(job, dict) else None + + +def _recordJobOutput(event): + """Synchronously record a finalized output file's id onto its owning job. + + Correlate the upload to a job by the finalized file's ACTUAL parent folder (a + File doc carries only ``itemId``, so this hops item -> folder), require the + reference ``identifier`` to be one the job DECLARED, and record the file id + keyed by that identifier (``otherFields`` dotted key -> Mongo ``$set`` nests + per identifier, so N outputs each bind without overwriting). Fail closed: a + foreign or uncorrelated upload, or an undeclared / unsafe identifier, is + ignored. This event fires synchronously before ``finalizeUpload`` returns, so + a DB failure while recording FAILS the upload (and the worker task) instead of + allowing a false SUCCESS with an unbound result. + """ + info = getattr(event, "info", None) + if not isinstance(info, dict): + return + upload = info.get("upload") + if not isinstance(upload, dict): + return + ref = _parseOutputReference(upload.get("reference")) + if ref is None: + return + fileDoc = info.get("file") + if not isinstance(fileDoc, dict): + return + fileId = fileDoc.get("_id") + itemId = fileDoc.get("itemId") + if not fileId or not itemId: + return + # A File doc has no ``folderId``; only its parent Item does. The item's parent + # folder is the job-correlation key. + item = Item().load(itemId, force=True, exc=False) + parentFolderId = item.get("folderId") if isinstance(item, dict) else None + if parentFolderId is None: + return + job = _jobForOutputFolder(parentFolderId) + if not isinstance(job, dict): + return + identifier = ref["identifier"] + if identifier not in _declaredOutputIdentifiers(job): + # An upload into the job's own folder whose identifier the job never + # declared is still refused -- correlation binds only declared outputs. + return + + girder_job.Job().updateJob( + job, + otherFields={"%s.%s" % (_OUTPUTS_FIELD, identifier): str(fileId)}, + ) + + +def _cascadeDeleteJobOwnedResources(event): + """``model.job.remove`` handler: enforce job/output ownership on deletion. + + For a job that owns an output folder (``_OUTPUT_FOLDER_ID_FIELD`` present): + + * REFUSE to remove a non-terminal job -- raise so Girder never reaches the + DB delete (this protects our own DELETE route AND Girder's built-in job + route and any direct ``Job.remove`` caller); + * otherwise cascade-delete the owned output folder (``Folder().remove`` + cascades to its items / subfolders / pending uploads) and then any + remaining staged input items; + * treat an already-missing owned folder as success (deletion is retryable); + * let a folder-removal failure PROPAGATE so the model retains the job record + -- the retained record stays the discoverable owner of whatever is left, + and a later retry completes the cascade. + + Jobs without the ownership field are untouched (standard removal proceeds). + """ + from .results import isTerminalStatus + + job = getattr(event, "info", None) + if not isinstance(job, dict): + return + folderId = job.get(_OUTPUT_FOLDER_ID_FIELD) + transientItemIds = job.get(TRANSIENT_STAGED_META_KEY) or [] + if not folderId and not transientItemIds: + return # not an owned job -- do not interfere with standard removal + if folderId: + if not isTerminalStatus(job.get("status")): + raise RestException( + "Cannot delete a job that is not finished", code=409 + ) + # The output folder is the critical owned resource (it holds the + # results); a failure here propagates so the job is retained and the + # delete is retryable. The in-progress marker (NOT a DB unset, which + # would break retryability on a partial failure) stops the + # model.folder.remove reverse cascade from re-entering Job.remove + # for this same job mid-delete. + folder = Folder().load(folderId, force=True, exc=False) + if folder is not None: + _CASCADING_FOLDER_IDS.add(str(folderId)) + try: + Folder().remove(folder) + finally: + _CASCADING_FOLDER_IDS.discard(str(folderId)) + # Staged inputs are transient (also orphan-swept); clean them best-effort so a + # stray input never blocks removing the (already-gone) results folder + job. + # Swept even when the folder pointer is absent: the reverse cascade unsets it + # before re-entering removal, and its jobs still own their staged inputs. + _removeTransientItems(transientItemIds) + + +# Output-folder ids currently being removed by the job-side cascade above. +# GIL-atomic set mutations; entries live only for the synchronous span of one +# Folder().remove call, so the reverse handler below can tell "the job is +# deleting its own folder" apart from "a user deleted the folder in Girder". +_CASCADING_FOLDER_IDS = set() + + +def _cascadeDeleteFolderOwnedJob(event): + """``model.folder.remove`` handler: removing a job's output folder removes the job. + + The reverse of ``_cascadeDeleteJobOwnedResources``: the private output + folder is the job's sole owned storage, so deleting it in the Girder + hierarchy means "delete this job". Removing the ``volview-jobs`` container + recurses through the per-job subfolders, firing this handler once per job. + + Mirrored invariants: + + * REFUSE to remove a live job's folder -- a non-terminal owned job raises, + aborting the folder removal (same guard as the job-side cascade); + * recursion guard -- unset ``volviewOutputFolderId`` on the job (DB + the + in-memory doc) BEFORE ``girder_job.Job().remove``, so the job-side cascade sees + no owned folder and only sweeps staged inputs. Unsetting is safe in this + direction: the folder is going away regardless, so a retained pointer + could only dangle. + + No-ops fail closed: an unmarked folder, a folder no job owns (the container, + a pre-publication orphan, an already-cascaded delete), or a removal driven + by the job-side cascade itself (``_CASCADING_FOLDER_IDS``). + """ + + from .results import isTerminalStatus + + folder = getattr(event, "info", None) + if not isinstance(folder, dict): + return + if not (folder.get("meta") or {}).get(JOB_OUTPUT_FOLDER_META_KEY): + return + folderId = folder.get("_id") + if folderId is None or str(folderId) in _CASCADING_FOLDER_IDS: + return + job = _jobForOutputFolder(folderId) + if job is None: + return + if not isTerminalStatus(job.get("status")): + raise RestException( + "Cannot delete the output folder of a job that is not finished; " + "cancel the job first", + code=409, + ) + girder_job.Job().update( + {"_id": job["_id"]}, {"$unset": {_OUTPUT_FOLDER_ID_FIELD: ""}} + ) + job.pop(_OUTPUT_FOLDER_ID_FIELD, None) + try: + girder_job.Job().remove(job) + except Exception: + # Restore the ownership pointer so a failed job removal never orphans + # the history row: the raise aborts the folder delete (the shell is + # retained), and the restored pointer keeps the job correlated with it + # for a later retry. + girder_job.Job().update( + {"_id": job["_id"]}, + {"$set": {_OUTPUT_FOLDER_ID_FIELD: str(folderId)}}, + ) + job[_OUTPUT_FOLDER_ID_FIELD] = str(folderId) + raise + + +def _folderChainMatchesTargets(folderId, folderTargets, baseTargets): + """Whether ``folderId`` is/descends from any folder in ``folderTargets``, + or its ROOT folder hangs directly under any ``(parentType, id)`` pair in + ``baseTargets`` (a collection or user about to be recursively deleted). + + Walks the folder's parent chain upward (owned output folders nest a couple + of levels deep, so the chain is short). Fail closed on a missing folder; + a (corrupt) parent cycle terminates via ``seen``. + """ + seen = set() + currentId = folderId + while currentId is not None and str(currentId) not in seen: + if str(currentId) in folderTargets: + return True + seen.add(str(currentId)) + folder = Folder().load(currentId, force=True, exc=False) + if not isinstance(folder, dict): + return False + if folder.get("parentCollection") != "folder": + return ( + folder.get("parentCollection"), + str(folder.get("parentId")), + ) in baseTargets + currentId = folder.get("parentId") + return False + + +def _liveJobOwningFolderUnderTargets(folderIds=(), baseParents=()): + """The first non-terminal job whose owned output folder is one of + ``folderIds``, a descendant of one, or contained (transitively) in any + ``(parentType, id)`` collection/user in ``baseParents`` — else ``None``. + + Persisted job ownership is the AUTHORITATIVE record: neither the deleted + folder nor anything on the path needs to carry the folder marker, so + deleting an unmarked ancestor (the launch folder, a project root) or a + folder whose marker was stripped cannot bypass the guard. Live owned jobs + are few at any moment, so checking each one's ancestor chain is cheap; the + query excludes settled history via the indexed ownership field + status. + """ + + from .results import terminalStatuses + + folderTargets = {str(folderId) for folderId in folderIds} + baseTargets = {(parentType, str(_id)) for parentType, _id in baseParents} + if not folderTargets and not baseTargets: + return None + jobs = girder_job.Job().find( + { + _OUTPUT_FOLDER_ID_FIELD: {"$exists": True}, + "status": {"$nin": list(terminalStatuses())}, + } + ) + for job in jobs: + if _folderChainMatchesTargets( + job.get(_OUTPUT_FOLDER_ID_FIELD), folderTargets, baseTargets + ): + return job + return None + + +def _refuseIfLiveJobUnder(folderIds=(), baseParents=()): + if _liveJobOwningFolderUnderTargets(folderIds, baseParents) is not None: + raise RestException( + "Cannot delete a processing job's output folder while the job is " + "still running; cancel the job first", + code=409, + ) + + +def _refuseLiveJobFolderRestDelete(event): + """``rest.delete.folder/:id.before`` guard: 409 a live job's folder delete. + + ``Folder.remove`` deletes contents (``clean``) BEFORE ``model.folder.remove`` + fires, so the reverse-cascade handler's non-terminal refusal can only save + the folder shell, not what was inside. This REST-level guard runs before the + delete handler touches anything, covering a live job's own folder, the + ``volview-jobs`` container, and ANY ancestor of either (deleting the launch + folder or a project root recursively cleans the job folder just the same). + Direct model callers bypass REST and still hit the (shell-level) late guard. + """ + info = getattr(event, "info", None) + folderId = (info or {}).get("id") + if folderId: + _refuseIfLiveJobUnder(folderIds=[folderId]) + + +def _refuseLiveJobCollectionRestDelete(event): + """``rest.delete.collection/:id.before``: the same preflight for collection + deletion, which recursively removes every folder inside without ever + passing through ``DELETE /folder/:id``.""" + info = getattr(event, "info", None) + collectionId = (info or {}).get("id") + if collectionId: + _refuseIfLiveJobUnder(baseParents=[("collection", collectionId)]) + + +def _refuseLiveJobUserRestDelete(event): + """``rest.delete.user/:id.before``: the same preflight for user deletion + (removes the user's whole folder tree).""" + info = getattr(event, "info", None) + userId = (info or {}).get("id") + if userId: + _refuseIfLiveJobUnder(baseParents=[("user", userId)]) + + +def _refuseLiveJobResourceRestDelete(event): + """``rest.delete.resource.before``: the same preflight for the batch + ``DELETE /resource`` route (arbitrary folder/collection/user ids). + + A malformed ``resources`` payload is left for the route itself to 400; + item ids are ignored (an item cannot contain a job's output folder). + """ + info = getattr(event, "info", None) + raw = ((info or {}).get("params") or {}).get("resources") + try: + resources = json.loads(raw) if isinstance(raw, str) else raw + except ValueError: + return + if not isinstance(resources, dict): + return + ids = { + model: [_id for _id in values if _id] + for model, values in resources.items() + if isinstance(values, list) + } + _refuseIfLiveJobUnder( + folderIds=ids.get("folder", ()), + baseParents=[ + (parentType, _id) + for parentType in ("collection", "user") + for _id in ids.get(parentType, ()) + ], + ) diff --git a/girder_volview/backend/results.py b/girder_volview/backend/results.py new file mode 100644 index 0000000..9882302 --- /dev/null +++ b/girder_volview/backend/results.py @@ -0,0 +1,427 @@ +"""Processing backend -- job status projection + result collection (the read path). + +Girder ``JobStatus`` projects to the contract's neutral shapes (never the girder +enum on the wire). Results come from the file ids recorded ON the job by +``outputs._recordJobOutput`` -- reference-bound, never a folder-name scan. +""" + +import functools + +from bson.objectid import ObjectId +from girder import logger +from girder_jobs.constants import JobStatus + +from ..utils import makeFileDownloadUrl, _toIso +from .config import processingProviderId +from .inputs import _LAUNCH_FOLDER_FIELD, _TASK_ID_FIELD, readableFilesById +from .outputs import ( + _OUTPUTS_FIELD, + _OUTPUT_SPECS_FIELD, + _declaredOutputIdentifiers, +) + + +@functools.cache +def _workerActiveStates(): + """girder_worker ``CustomJobStatus`` active-state codes (with numeric fallback). + + Core's ``JobStatus`` map has no entry for these, so without them a running job + would default to ``"pending"`` and a polling client would see it REGRESS. The + numeric literals are the stable wire integers used when girder_worker (an + optional runtime dependency) is not importable. + """ + try: + from girder_worker.utils import CustomJobStatus + + return { + CustomJobStatus.FETCHING_INPUT, + CustomJobStatus.CONVERTING_INPUT, + CustomJobStatus.CONVERTING_OUTPUT, + CustomJobStatus.PUSHING_OUTPUT, + CustomJobStatus.CANCELING, + } + except Exception: + return { + 820, # CustomJobStatus.FETCHING_INPUT + 821, # CustomJobStatus.CONVERTING_INPUT + 822, # CustomJobStatus.CONVERTING_OUTPUT + 823, # CustomJobStatus.PUSHING_OUTPUT + 824, # CustomJobStatus.CANCELING + } + + +@functools.cache +def _jobStateMap(): + """The girder ``JobStatus`` -> neutral projected-state map, built once.""" + + return { + JobStatus.INACTIVE: "pending", + JobStatus.QUEUED: "pending", + JobStatus.RUNNING: "running", + JobStatus.SUCCESS: "success", + JobStatus.ERROR: "error", + JobStatus.CANCELED: "cancelled", + } + + +def isTerminalStatus(status): + """Whether a Girder ``JobStatus`` is terminal (SUCCESS / ERROR / CANCELED). + + The single definition of "the job has settled": the terminal-time scan, the + ownership deletion guard, the transient-input cleanup, and the DELETE route + all read it, so what counts as terminal cannot drift between them. + """ + return status in terminalStatuses() + + +@functools.cache +def terminalStatuses(): + """The terminal ``JobStatus`` set itself (for Mongo ``$nin`` queries).""" + + return frozenset({JobStatus.SUCCESS, JobStatus.ERROR, JobStatus.CANCELED}) + + +def _projectJobState(job): + """The neutral projected job state (a ``jobStateSchema`` value) from Girder's + ``JobStatus``. + + The single shared JobStatus->state map, reached through ``_projectJobFacts`` + (and ``_readableOutputFilesForJobs``), so the status and history reads cannot + disagree about execution state. Neutral names only — never the girder + ``JobStatus`` enum on the wire. + An unknown status maps to ``"pending"`` (fail closed), except girder_worker's + active states, which project to ``"running"`` so an active job never regresses. + Output publication never changes this execution state; ``_projectJobFacts`` + carries result readiness separately. + """ + status = job.get("status") + state = _jobStateMap().get(status) + if state is not None: + return state + if status in _workerActiveStates(): + return "running" + return "pending" + + +def _progressRatio(job): + """The job's clamped ``[0, 1]`` progress ratio, or ``None`` when unavailable. + + Shared by the status projection and the history summary so the summary never + rebuilds the full status projection (errorTail log join included) just to read + progress. The clamp is load-bearing: a worker/CLI reporting >100% (or a + negative) fails the client's ``min(0).max(1)`` history-page schema and makes it + reject the WHOLE page, losing re-discovery for every job in it. + """ + progress = job.get("progress") or {} + if not progress.get("total") or progress.get("current") is None: + return None + try: + ratio = float(progress["current"]) / float(progress["total"]) + except (TypeError, ValueError, ZeroDivisionError): + # ValueError: updateJob stores whatever a writer PUT — a non-numeric + # current/total string must not 500 the whole history page. + return None + return min(1.0, max(0.0, ratio)) + + +def _projectJobStatus(job, user=None): + """Convert Girder Job status to ProcessingJobStatus.""" + facts = _projectJobFacts(job, user) + state = facts["state"] + out = { + "jobId": str(job["_id"]), + "state": state, + "resultState": facts["resultState"], + } + if state == "error": + log = job.get("log") or [] + if isinstance(log, list): + tail = "".join(log[-20:]) + else: + tail = str(log)[-2000:] + out["errorTail"] = tail + progress = _progressRatio(job) + if progress is not None: + out["progress"] = progress + return out + + +def _transitionTime(job, status): + for transition in job.get("timestamps") or []: + if isinstance(transition, dict) and transition.get("status") == status: + return transition.get("time") + return None + + +def _terminalTime(job): + """The job's most recent terminal transition time, or ``None``.""" + for transition in reversed(job.get("timestamps") or []): + if isinstance(transition, dict) and isTerminalStatus(transition.get("status")): + return transition.get("time") + return None + + +def _outputSummary(job, user, facts=None): + """Neutral output-health counts for the job-history wire shape. + + - ``recorded``: declared outputs that recorded AND resolve to a readable file. + - ``missing``: settled declared outputs that never recorded, plus recorded + ids whose file is gone/unreadable. + """ + facts = facts or _projectJobFacts(job, user) + return { + "recorded": len(facts["resolved"]), + "missing": facts["missing"], + } + + +def _projectJobHistorySummary(job, user, readableOutputFiles=None): + """Project one Girder job into the lightweight history wire shape.""" + + creatorName = ( + " ".join( + filter( + None, + [ + user.get("firstName"), + user.get("lastName"), + ], + ) + ) + or user.get("login") + or str(user.get("_id") or "") + ) + facts = _projectJobFacts(job, user, readableOutputFiles=readableOutputFiles) + summary = { + "jobId": str(job["_id"]), + "taskId": str(job.get(_TASK_ID_FIELD) or ""), + "taskTitle": str(job.get("title") or job.get(_TASK_ID_FIELD) or ""), + "createdBy": {"id": str(user["_id"]), "name": creatorName}, + "createdAt": _toIso(job.get("created")) or "", + "state": facts["state"], + "resultState": facts["resultState"], + "outputSummary": _outputSummary(job, user, facts), + } + startedAt = _toIso(_transitionTime(job, JobStatus.RUNNING)) + finishedAt = _toIso(_terminalTime(job)) + if startedAt: + summary["startedAt"] = startedAt + if finishedAt: + summary["finishedAt"] = finishedAt + progress = _progressRatio(job) + if progress is not None: + summary["progress"] = progress + return summary + + +def _intentForOutput(out, url, name, providerId, jobId): + """Build the declarative result intent for one output. + + Results cross the wire as declarative intents the client's single applier + applies — never a ``role`` the client switches on. The vocabulary the client + validates (VolView ``backend-contract/processing/wire.ts``): a labelmap → + ``add-segment-group``, a plain image → ``add-base-image``. Any other file + remains an ordinary result record with no state directive. + + A labelmap intent carries a provider-qualified + ``source: {providerId, jobId, outputId}`` provenance tag (``outputId`` = the + CLI's output identifier) so the idempotency key remains unique when two + providers use the same raw job/output ids and round-trips the + ``.volview.zip``. A labelmap's segment names/colors travel *inside* the + ``.seg.nrrd`` file as embedded metadata and are read client-side, so the + backend sets no ``segments`` payload; the wire field stays optional. + Validates against the contract ``result-intent`` schema. + """ + fileRef = {"url": url, "name": name} + if out["isLabel"]: + return { + "intent": "add-segment-group", + **fileRef, + "source": { + "providerId": str(providerId), + "jobId": str(jobId), + "outputId": out["name"], + }, + } + if out["tag"] == "image": + return {"intent": "add-base-image", **fileRef} + return fileRef + + +def _recordedJobOutputs(job): + """The ``{identifier: fileId}`` map the upload-finalization handler + recorded (or {}).""" + outputs = (job or {}).get(_OUTPUTS_FIELD) + return dict(outputs) if isinstance(outputs, dict) else {} + + +def _recordedOutputSpecs(job): + """The declared output specs recorded at submit (or []).""" + specs = (job or {}).get(_OUTPUT_SPECS_FIELD) + return list(specs) if isinstance(specs, list) else [] + + +def _projectJobFacts(job, user, readableOutputFiles=None): + """Project the canonical execution and output-readiness facts for one job. + + Status, history, and result reads all consume this projection so execution + state never changes to hide output publication, and missing-output accounting + cannot disagree between endpoints. File readability comes from the ONE + batched loader (``_readableOutputFilesForJobs``): the history page passes its + page-wide map in, single-job callers omit it and the loader runs for just this + job — either way the same two-query ACL check decides readability. + """ + state = _projectJobState(job) + if state in {"pending", "running"}: + return { + "state": state, + "resultState": "waiting", + "resolved": [], + "missing": 0, + } + if state in {"error", "cancelled"}: + return { + "state": state, + "resultState": "unavailable", + "resolved": [], + "missing": 0, + } + + specs = { + spec["name"]: spec + for spec in _recordedOutputSpecs(job) + if isinstance(spec, dict) and isinstance(spec.get("name"), str) + } + recorded = _recordedJobOutputs(job) + if readableOutputFiles is None: + readableOutputFiles = _readableOutputFilesForJobs([job], user) + resolved = [] + unreadable = 0 + for outputId, spec in specs.items(): + fileId = recorded.get(outputId) + if fileId is None: + continue + fileDoc = readableOutputFiles.get(str(fileId)) + if fileDoc is None: + unreadable += 1 + else: + resolved.append({"out": spec, "fileDoc": fileDoc}) + + unrecorded = len(set(specs) - set(recorded)) + missing = unrecorded + unreadable + return { + "state": state, + "resultState": "incomplete" if missing else "ready", + "resolved": resolved, + "missing": missing, + } + + +def _readableOutputFilesForJobs(jobs, user): + """Load all readable declared output files for a set of jobs in two queries. + + The ONE loading path deciding output readability — the history page passes + its whole page, and ``_projectJobFacts`` routes single-job status/result + reads through it too, so the ACL semantics cannot drift between endpoints. + The recorded/missing counts are readability-aware, so the ACL check is + load-bearing; ``inputs.readableFilesById`` is the shared batched boundary. + Files with invalid ids, missing parents, or unreadable parents are absent + from the map and therefore count as missing. The returned map is keyed by + string id because persisted output ids and model documents may use different + ObjectId/string representations. The projection carries + ``name``/``mimeType``/``size`` because ``_collectJobResults`` builds result + intents (download url + file metadata) from these same docs. + """ + fileIdsByString = {} + for job in jobs: + if _projectJobState(job) != "success": + continue + declared = _declaredOutputIdentifiers(job) + for outputId, fileId in _recordedJobOutputs(job).items(): + if outputId not in declared or fileId is None: + continue + try: + objectId = fileId if isinstance(fileId, ObjectId) else ObjectId(fileId) + except (TypeError, ValueError): + continue + fileIdsByString.setdefault(str(objectId), objectId) + if not fileIdsByString: + return {} + return readableFilesById( + fileIdsByString.values(), + user, + fields={"_id": 1, "itemId": 1, "name": 1, "mimeType": 1, "size": 1}, + ) + + +def _collectJobResults(job, user, facts=None): + """Resolve a job's outputs to declarative result intents — reference-bound. + + Reads the ``{identifier: fileId}`` map ``outputs._recordJobOutput`` recorded ON + the job (never a folder-name scan, never ``_original_params``), resolves the + files through the batched READ-permission loader, and projects each into its + result intent with a ``makeFileDownloadUrl`` download url — origin-relative and + filename-encoded, so non-default API mounts work. Returns ``(results, + missing)`` where ``missing`` counts settled unrecorded outputs plus recorded + outputs whose file is gone/unreadable — loss is countable, never a silently + shorter list. Two concurrent same-name jobs can never cross results: each reads + only the ids bound to itself. + """ + facts = facts or _projectJobFacts(job, user) + resolved = facts["resolved"] + + # The wire shape is the intent object itself — `{intent, url, name, source?}` + # — plus the `id`/`mimeType`/`size` file metadata the client's JobList reads. + providerId = processingProviderId(job[_LAUNCH_FOLDER_FIELD]) + results = [] + for entry in resolved: + out = entry["out"] + fileDoc = entry["fileDoc"] + url = makeFileDownloadUrl(fileDoc) + intent = _intentForOutput( + out, url, fileDoc["name"], providerId, job["_id"] + ) + result = { + **intent, + "id": str(fileDoc["_id"]), + "mimeType": fileDoc.get("mimeType"), + "size": fileDoc.get("size"), + } + results.append(result) + return results, facts["missing"] + + +def _jobResultsPayload(job, user): + """Apply honest result-read semantics and return the wire result envelope. + + Waiting and unavailable results return the typed conflict body the + route serves as HTTP 409. Ready and incomplete reads return HTTP 200; partial + and total output loss both use an incomplete envelope with an accurate count. + """ + facts = _projectJobFacts(job, user) + state = facts["state"] + resultState = facts["resultState"] + if resultState == "unavailable": + return { + "code": "results_unavailable", + "message": "Job %s results are unavailable (state=%s)" + % (job.get("_id"), state), + "state": state, + "resultState": resultState, + } + if resultState == "waiting": + return { + "code": "results_not_ready", + "message": "Job %s results are not ready (resultState=%s)" + % (job.get("_id"), resultState), + "state": state, + "resultState": resultState, + } + results, missing = _collectJobResults(job, user, facts) + if missing: + logger.info( + "[volview_processing] job %s: %d declared output(s) missing", + job.get("_id"), + missing, + ) + return {"resultState": resultState, "intents": results, "missing": missing} diff --git a/girder_volview/backend/routes.py b/girder_volview/backend/routes.py new file mode 100644 index 0000000..63abcbb --- /dev/null +++ b/girder_volview/backend/routes.py @@ -0,0 +1,918 @@ +"""Processing backend -- REST routes + job creation + route registration. + +Cross-module helper calls are MODULE-QUALIFIED on purpose (``submit._foo`` / +``inputs._foo`` / ``outputs._foo`` / ``results._foo``) so a test that patches a +helper on its DEFINING module reaches the call site here; a bare +``from .submit import _foo`` binds a name ``setattr(submit, "_foo", ...)`` +cannot reach. +""" + +import base64 +import binascii +import copy +import datetime +import json +import threading +import time +import uuid + +import cherrypy +from girder import events, logger +from girder.api import access +from girder.api.describe import Description, autoDescribeRoute +from girder.api.rest import Resource, boundHandler +from girder.constants import AccessType, SortDir, TokenScope +from girder.exceptions import RestException, ValidationException +from girder.models.folder import Folder + +# Module-object import (not ``from ... import Job``): call sites resolve +# ``girder_job.Job`` at call time, so tests may monkeypatch the class on +# ``girder_jobs.models.job`` and be seen here. +from girder_jobs.models import job as girder_job + +from ..utils import ( + _toIso, + makeFileDownloadUrl, + JOB_OUTPUT_FOLDER_META_KEY, + TRANSIENT_STAGED_META_KEY, +) +from .config import PROCESSING_ROUTE_NAME +from .slicer_spec import declared_params, translate_slicer_xml, validate_task_spec +from . import inputs, submit, outputs, results + + +@access.public(cookie=True, scope=TokenScope.DATA_READ) +@boundHandler +@autoDescribeRoute( + Description("List processing tasks available for a folder.") + .modelParam("folderId", model=Folder, level=AccessType.READ) + .produces(["application/json"]) +) +def listTasks(self, folder): + user = self.getCurrentUser() + tasks = [] + if user and submit._slicerCliAvailable(): + try: + tasks.extend( + [submit._cliItemToSummary(c) for c in submit._scopedCliItems(user)] + ) + except Exception: + logger.exception("Failed to list slicer_cli_web items") + return tasks + + +# Explicit short lifetime for the CLI-container token. Without ``days`` Girder +# applies ``core.cookie_lifetime`` (180 days by default), leaving a broad +# data-plane credential valid for months after the job ends. One day covers queue +# wait + run time and bounds exposure if the token leaks (a compromised image, +# broker, or command-line capture). +_CONTAINER_TOKEN_TTL_DAYS = 1.0 + +JOB_HISTORY_PAGE_DEFAULT = 25 +JOB_HISTORY_PAGE_MAX = 100 +JOB_HISTORY_INDEX = "volview_job_history" +JOB_OUTPUT_FOLDER_INDEX = "volview_output_folder" +_SUBMISSION_ID_FIELD = "volviewSubmissionId" +_SUBMITTED_PARAMETERS_FIELD = "volviewSubmittedParameters" + + +def ensureJobHistoryIndexes(jobModel=None): + """Install the indexes the history list and output-correlation queries need. + + A compound index backs the personal newest-first history page. A point-lookup + index on the private output-folder id backs ``outputs._jobForOutputFolder``, + which runs a ``findOne`` for EVERY finalized output upload -- without it each + correlation is a full jobs-collection scan that worsens as history grows. + """ + if jobModel is None: + jobModel = girder_job.Job() + jobModel.collection.create_index( + [ + (inputs._LAUNCH_FOLDER_FIELD, 1), + ("userId", 1), + ("created", -1), + ("_id", -1), + ], + name=JOB_HISTORY_INDEX, + ) + jobModel.collection.create_index( + [(outputs._OUTPUT_FOLDER_ID_FIELD, 1)], + name=JOB_OUTPUT_FOLDER_INDEX, + ) + + +def _encodeJobCursor(job): + payload = json.dumps( + { + "created": _toIso(job.get("created")), + "id": str(job["_id"]), + }, + separators=(",", ":"), + ).encode("utf8") + return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=") + + +def _decodeJobCursor(cursor): + from bson.objectid import ObjectId + + try: + padded = cursor + "=" * (-len(cursor) % 4) + value = json.loads(base64.urlsafe_b64decode(padded).decode("utf8")) + created = datetime.datetime.fromisoformat(value["created"]) + if created.tzinfo is not None: + created = created.astimezone(datetime.timezone.utc).replace(tzinfo=None) + return created, ObjectId(value["id"]) + except (TypeError, ValueError, KeyError, json.JSONDecodeError, binascii.Error): + raise RestException("Invalid job history cursor", code=400) from None + + +def _jobHistoryPageSize(limit): + try: + pageSize = int(limit if limit is not None else JOB_HISTORY_PAGE_DEFAULT) + if pageSize < 1 or pageSize > JOB_HISTORY_PAGE_MAX: + raise ValueError() + return pageSize + except (TypeError, ValueError): + raise RestException("Invalid job history limit", code=400) from None + + +def _jobCursorContinuation(cursor): + created, jobId = _decodeJobCursor(cursor) + return [ + {"created": {"$lt": created}}, + {"created": created, "_id": {"$lt": jobId}}, + ] + + +@access.public(cookie=True, scope=TokenScope.DATA_READ) +@boundHandler +@autoDescribeRoute( + Description("List the current user's complete processing-job history.") + .notes( + "Returns a bounded newest-first page of lightweight summaries. The " + "continuation cursor is opaque; pagination bounds responses, never " + "history retention. Logs and submitted parameters are detail-only." + ) + .modelParam("folderId", model=Folder, level=AccessType.READ) + .param("limit", "Page size (1-100).", required=False, dataType="integer") + .param("cursor", "Opaque continuation cursor.", required=False) + .produces(["application/json"]) +) +def listJobHistory(self, folder, limit=JOB_HISTORY_PAGE_DEFAULT, cursor=None): + user = self.getCurrentUser() + if not user: + return {"jobs": [], "nextCursor": None} + + pageSize = _jobHistoryPageSize(limit) + query = { + inputs._LAUNCH_FOLDER_FIELD: str(folder["_id"]), + "userId": user["_id"], + } + if cursor: + query["$or"] = _jobCursorContinuation(cursor) + found = girder_job.Job().findWithPermissions( + query=query, + user=user, + jobUser=user, + level=AccessType.READ, + sort=[("created", SortDir.DESCENDING), ("_id", SortDir.DESCENDING)], + limit=pageSize + 1, + # The summary projection never reads the log, which is unbounded (multi-MB + # on chatty/failed CLIs), so exclude it from every page. Mirrors + # Job.load(includeLog=False)'s {'log': False} projection. + fields={"log": False}, + ) + page = list(found) + hasMore = len(page) > pageSize + page = page[:pageSize] + readableOutputFiles = results._readableOutputFilesForJobs(page, user) + return { + "jobs": [ + results._projectJobHistorySummary( + job, user, readableOutputFiles=readableOutputFiles + ) + for job in page + ], + "nextCursor": _encodeJobCursor(page[-1]) if hasMore else None, + } + + +@access.public(cookie=True, scope=TokenScope.DATA_READ) +@boundHandler +@autoDescribeRoute( + Description("Get the VolView task spec for a task.") + .modelParam("folderId", model=Folder, level=AccessType.READ) + .param("taskId", "The task identifier.", paramType="path") +) +def getTaskSpec(self, folder, taskId): + # The Slicer XML is translated into VolView's task spec server-side, so the + # client never parses backend XML. + user = self.getCurrentUser() + if not submit._slicerCliAvailable(): + raise RestException("slicer_cli_web is not installed", code=404) + scoped = submit._findScopedCliItem(taskId, user) + if not scoped: + raise RestException("Unknown taskId", code=404) + # translate_slicer_xml needs the strict parse (title/description + # + ordered params), which ``parse_cli`` does not carry, so it parses the XML + # itself; the scoped parse is consumed by runTask. + cliItem, _parsedCli = scoped + try: + return validate_task_spec(translate_slicer_xml(cliItem.xml, str(taskId))) + except ValueError as exc: + logger.error("Invalid VolView task spec for task %s: %s", taskId, exc) + raise RestException("Task specification is invalid", code=500) from None + + +# The single server-owned container every per-job output folder nests inside: one +# hierarchy entry per launch folder no matter how many jobs accumulate, and one +# ADMIN-gated "clear this dataset's job history" gesture (removing it recurses +# through the per-job folders, firing the reverse cascade per job). The name is +# reserved: a pre-existing USER folder with this name is never adopted -- reuse is +# gated on the server-owned marker, and an unmarked name collision refuses the +# submission (409) instead. +JOBS_CONTAINER_NAME = "volview-jobs" + + +def _jobsContainerFolder(launchFolder, user): + """Create-or-reuse the launch folder's server-owned ``volview-jobs`` container. + + Reuse requires the ``volviewJobOutputFolder`` marker -- the server-owned + identity stamped at creation. Adopting a user's pre-existing folder that + merely shares the reserved name would silently hide its contents from launch + manifests and turn the container-delete gesture into "delete unrelated user + data", so an unmarked collision refuses the submission with a clear 409. The + marker is stamped ONLY on a folder this call itself created: ``createFolder`` + never reuses (name-collision raises ``ValidationException``), so a folder + someone else made in the check-create window re-runs the marker check instead + of being adopted. Create and stamp are two writes, so an unmarked collision + gets a short grace period (a concurrent submission's container between its + create and its stamp) before the 409, and a failed stamp removes the created + folder rather than leave an unmarked container that would 409 every future + submission. Its ACL is the launch folder's (copied by ``createFolder``): + collaborators may see the container, but each per-job folder inside keeps its + submitter-only ACL. + """ + + def findContainer(): + return Folder().findOne( + { + "parentId": launchFolder["_id"], + "parentCollection": "folder", + "name": JOBS_CONTAINER_NAME, + } + ) + + def isMarked(folderDoc): + return bool((folderDoc.get("meta") or {}).get(JOB_OUTPUT_FOLDER_META_KEY)) + + def awaitMarkedContainer(): + """The existing marked container, ``None`` when absent, or a 409 for an + unmarked collision that outlasts the create->stamp grace period.""" + for _ in range(10): + existing = findContainer() + if existing is None or isMarked(existing): + return existing + time.sleep(0.05) + raise RestException( + "A folder named '%s' already exists here and is not a processing " + "jobs container; rename or remove it to run processing tasks" + % JOBS_CONTAINER_NAME, + code=409, + ) + + container = awaitMarkedContainer() + if container is not None: + return container + try: + created = Folder().createFolder( + parent=launchFolder, + name=JOBS_CONTAINER_NAME, + parentType="folder", + creator=user, + public=False, + ) + except ValidationException: + # Lost a creation race: a same-named sibling appeared between the check + # and the create. Re-run the marker check -- a concurrent submission's + # container is reused; a user's folder 409s (never adopted). + container = awaitMarkedContainer() + if container is None: + raise + return container + try: + return Folder().setMetadata(created, {JOB_OUTPUT_FOLDER_META_KEY: True}) + except Exception: + Folder().remove(created) + raise + + +def _createJobOutputFolder(launchFolder, user, submissionId): + """Create the job's private, server-owned output folder for a submission. + + Lives inside the launch folder's ``volview-jobs`` container. Every + declared output is forced into this folder, and it is the SOLE + output-correlation + ownership key. Two steps make it private: + + * mark it ``volviewJobOutputFolder`` so the launch manifest excludes it and + its contents (job results take the job path only, never ordinary launch + data); + * REPLACE its ACL with a submitter-only ADMIN list. ``createFolder`` copies + the parent's ACL (``copyAccessPolicies``), which would otherwise leave + every launch-folder collaborator able to read the private results; + ``setAccessList(..., force=True, setPublic=False)`` strips that. Girder + system administrators keep their normal force access. + """ + created = Folder().createFolder( + parent=_jobsContainerFolder(launchFolder, user), + name="volview-job-%s" % submissionId, + parentType="folder", + creator=user, + public=False, + reuseExisting=False, + ) + try: + folder = Folder().setMetadata(created, {JOB_OUTPUT_FOLDER_META_KEY: True}) + Folder().setAccessList( + folder, + { + "users": [{"id": user["_id"], "level": AccessType.ADMIN}], + "groups": [], + }, + save=True, + force=True, + setPublic=False, + ) + return folder + except Exception: + try: + Folder().remove(created) + except Exception: + logger.exception( + "Failed to remove partially initialized job output folder %s", + created.get("_id"), + ) + raise + + +def _removeJobOutputFolder(folder): + """Best-effort removal of a pre-publication output folder (no job yet). + + Only called when a submission failed BEFORE any job was created, so no + ownership record exists to drive the normal deletion cascade. An + already-missing folder is a no-op. + """ + if not folder: + return + try: + Folder().remove(folder) + except Exception: + logger.exception( + "Failed to remove orphaned job output folder %s", folder.get("_id") + ) + + +def _requestCliItem(cliItem, initialJobFields): + """Copy a catalog CLI item and inject request-local initial job fields.""" + requestItem = copy.copy(cliItem) + requestItem.item = copy.deepcopy(getattr(cliItem, "item", {}) or {}) + meta = requestItem.item.setdefault("meta", {}) + dockerParams = meta.setdefault("docker-params", {}) + if not isinstance(dockerParams, dict): + raise ValidationException("CLI docker parameters are malformed") + catalogFields = dockerParams.get("girder_job_other_fields") or {} + if not isinstance(catalogFields, dict): + raise ValidationException("CLI initial job fields are malformed") + merged = dict(catalogFields) + merged.update(initialJobFields) + dockerParams["girder_job_other_fields"] = merged + return requestItem + + +def _genDockerJob(cliItem, params, user, initialJobFields): + """Create the slicer_cli_web docker job for a CLI item and return its doc. + + Isolated as the single live slicer_cli_web touch point so ``runTask`` (and + its tests) can drive job creation without the optional dependency. + """ + from girder.models.token import Token + from slicer_cli_web.rest_slicer_cli import genHandlerToRunDockerCLI + + # Scope-limit the container token to the data plane the CLI actually needs: + # read its inputs, write its outputs. Narrower than the ecosystem norm + # (slicer_cli_web mints full-auth tokens) without weakening Girder ACLs — the + # submitter's own read/write reach still bounds it. It is NOT persisted on the + # job or used as an ownership/correlation key (outputs bind by their private + # parent folder, never by a token). + token = Token().createToken( + user=user, + scope=[TokenScope.DATA_READ, TokenScope.DATA_WRITE], + days=_CONTAINER_TOKEN_TTL_DAYS, + ) + requestItem = _requestCliItem(cliItem, initialJobFields) + handler = genHandlerToRunDockerCLI(requestItem) + # slicer_cli_web only substitutes its GirderApiUrl()/GirderToken() runtime + # transforms when these keys are present in the params it processes + # (prepare_task `_add_optional_input_param` skips a param absent from args). + # The REST route defaults them in, but this backend calls `subHandler` + # directly. Empty -> the transforms substitute, and GirderToken resolves to + # THIS scoped token, not a broader one. Without them a CLI that fetches its own + # inputs by id (`reference="_girder_id_"`, e.g. a multi-file DICOM series) has + # no way to reach Girder. Harmless for a CLI that declares neither -- + # slicer_cli_web ignores undeclared args. + params = dict(params) + params.setdefault("girderApiUrl", "") + params.setdefault("girderToken", "") + # Take a copy so the handler can mutate freely. + job_obj = handler.subHandler(requestItem, copy.deepcopy(params), user, token) + job = job_obj.job if hasattr(job_obj, "job") else job_obj + return job + + +def _prepareSubmissionFields( + submissionId, folder, taskId, values, outputSpecs, transientItemIds, outputFolder +): + """Build every job association field before task publication. + + Includes the job's private output-folder id (``_OUTPUT_FOLDER_ID_FIELD``) — + the sole output-correlation + ownership key — so the folder id is part of the + FIRST job insert, queryable before any worker upload can race in. + + ``outputSpecs`` is the ``slicer_spec.parse_cli`` output descriptor list + ``runTask`` already parsed, threaded in rather than re-parsed here. + """ + fields = { + _SUBMISSION_ID_FIELD: submissionId, + inputs._LAUNCH_FOLDER_FIELD: str(folder["_id"]), + inputs._TASK_ID_FIELD: str(taskId), + outputs._OUTPUT_SPECS_FIELD: outputSpecs, + outputs._OUTPUT_FOLDER_ID_FIELD: str(outputFolder["_id"]), + outputs._OUTPUTS_FIELD: {}, + _SUBMITTED_PARAMETERS_FIELD: copy.deepcopy(values), + } + if transientItemIds: + fields[TRANSIENT_STAGED_META_KEY] = list(transientItemIds) + return fields + + +def _jobForSubmission(submissionId): + + return girder_job.Job().findOne({_SUBMISSION_ID_FIELD: submissionId}) + + +@access.public(cookie=True, scope=TokenScope.DATA_WRITE) +@boundHandler +@autoDescribeRoute( + Description("Submit a processing task.") + .modelParam("folderId", model=Folder, level=AccessType.WRITE) + .param("taskId", "The task identifier.", paramType="path") + .jsonParam( + "body", + "Submission payload: { values: { paramName: ProcessingValue, ... } }", + paramType="body", + required=False, + ) +) +def runTask(self, folder, taskId, body): + user = self.getCurrentUser() + values = (body or {}).get("values", {}) if isinstance(body, dict) else {} + if not isinstance(values, dict): + raise RestException("values must be an object of parameter values", code=400) + + # Reject a payload carrying reserved credentials before any task lookup or + # work; declaration-aware screens run after the CLI XML is parsed below. + submit._rejectReservedSubmitParams(values) + + if not submit._slicerCliAvailable(): + raise RestException("slicer_cli_web is not installed", code=500) + + scoped = submit._findScopedCliItem(taskId, user) + if not scoped: + raise RestException("Unknown taskId", code=404) + cliItem, parsedCli = scoped + + # The submission parses the CLI XML exactly twice: ``_findScopedCliItem`` + # already ran ``parse_cli``, whose ``outputs`` are reused here, and + # ``declared_params`` supplies the label-independent key/value declaration the + # grouped walk can't. Downstream guard/translate steps read these structures. + declared = declared_params(cliItem.xml) + outputSpecs = parsedCli["outputs"] + + # Screens the RAW client keys, so it must run before autofill adds + # server-owned output structures. A synthesized-folder collision, an + # undeclared key, an out-of-declaration value, or a missing required input + # is a boundary 400 naming the parameter, not a later job failure. + submit._rejectSynthesizedFolderParams(values, declared) + submit._rejectUndeclaredSubmitParams(values, declared) + submit._validateDeclaredSubmitValues(values, declared) + submit._rejectMissingRequiredParams(values, declared) + + # Auto-generate a deterministic output filename for any output param the user + # didn't fill (input file + CLI name + parameter name + extension). Names need + # not be unique: outputs bind to the job by its private output folder, not by + # name, so a duplicate filename can never cross results. + values = submit._autofillOutputs(dict(values), outputSpecs, cliItem.name) + + # The output folder is created BEFORE translating params or publishing the + # task: every declared output is forced into it, and its id is part of the + # first job insert so it is queryable before any worker upload can race in. + submissionId = uuid.uuid4().hex + outputFolder = _createJobOutputFolder(folder, user, submissionId) + + transientItemIds = [] + try: + # Resolves each bound input once (own-scheme validation + per-user ACL + # re-check), forces every declared output into the private output folder, + # and rejects a client-supplied folderRef. The authorized file documents + # are reused for transient detection so each URI's ACL check runs once. + params, resolvedInputFiles = submit._translateValuesToSlicerParams( + values, user, outputFolder, declared + ) + # Per-job input ownership: any staged (transient) input is COPIED into the + # job's private folder and the CLI params are rewritten onto the copies. + # The copies are recorded on the job so + # inputs._cleanupTransientOnJobDone deletes them at terminal state; the + # shared staged original is never a job dependency. + params, transientItemIds = inputs.copyStagedInputsIntoJobFolder( + params, resolvedInputFiles, user, outputFolder + ) + # INFO carries only routing identity; the translated CLI params can hold + # sensitive string values, so they stay at debug. + logger.info( + "[volview_processing] runTask folder=%s task=%s submission=%s", + folder["_id"], + taskId, + submissionId, + ) + logger.debug("[volview_processing] runTask params=%s", params) + + initialFields = _prepareSubmissionFields( + submissionId, + folder, + taskId, + values, + outputSpecs, + transientItemIds, + outputFolder, + ) + job_doc = _genDockerJob(cliItem, params, user, initialFields) + except Exception: + # run.delay can fail after Girder Worker's before_task_publish handler + # inserted the job. Resolve that ambiguity by the server-minted id. + job_doc = _jobForSubmission(submissionId) + if job_doc is None: + # No job exists (including a folderRef-rejection 400 before any work): + # remove the pre-publication output folder and staged inputs. + _removeJobOutputFolder(outputFolder) + inputs._removeTransientItems(transientItemIds) + else: + # A job WAS created: cancel it but RETAIN its ownership record so the + # normal terminal + deletion cascade cleans the output folder safely. + try: + + girder_job.Job().cancelJob(job_doc) + except Exception: + logger.exception( + "Failed to cancel ambiguously published job %s", + job_doc.get("_id"), + ) + raise + return {"jobId": str(job_doc["_id"])} + + +# Job-addressed routes are keyed by job id alone and gated by the job's OWN ACL. +# The launch folder is not part of a job's identity, so these carry no +# ``folderId``; they live on the folder-free ``volview_processing`` resource +# below. getJob / getJobResults are READ-gated; cancel and delete are WRITE-gated +# so a read-only viewer who can see a job's status cannot cancel or delete it. + + +def _loadJobForStatusProjection(jobId, user): + """Load a job for status projection, WITHOUT its log unless it is needed. + + The client polls status every ~2s per live job and the job log grows + unbounded, so the common load excludes it at the Mongo projection level + (``includeLog`` defaults False). Only the terminal-error projection reads it + (a bounded tail in ``results._projectJobStatus``), and the poller stops at + terminal, so the log is reloaded at most once per job. The detail route is + the full-log path. + """ + + job = girder_job.Job().load(jobId, user=user, level=AccessType.READ, exc=True) + if results._projectJobState(job) == "error": + job = girder_job.Job().load( + jobId, + user=user, + level=AccessType.READ, + exc=True, + includeLog=True, + ) + return job + + +@access.public(cookie=True, scope=TokenScope.DATA_READ) +@boundHandler +@autoDescribeRoute( + Description("Get job status.") + .param("jobId", "The job identifier.", paramType="path") + .produces(["application/json"]) +) +def getJob(self, jobId): + user = self.getCurrentUser() + job = _loadJobForStatusProjection(jobId, user) + return results._projectJobStatus(job, user) + + +@access.public(cookie=True, scope=TokenScope.DATA_READ) +@boundHandler +@autoDescribeRoute( + Description("Get detail-only job logs and submitted parameters.") + .param("jobId", "The job identifier.", paramType="path") + .produces(["application/json"]) +) +def getJobHistoryDetail(self, jobId): + user = self.getCurrentUser() + + job = girder_job.Job().load( + jobId, + user=user, + level=AccessType.READ, + exc=True, + includeLog=True, + ) + log = job.get("log") or [] + if not isinstance(log, list): + log = [str(log)] + parameters = job.get(_SUBMITTED_PARAMETERS_FIELD) or {} + if not isinstance(parameters, dict): + parameters = {} + return { + "jobId": str(job["_id"]), + "log": [str(line) for line in log], + "parameters": parameters, + } + + +@access.public(cookie=True, scope=TokenScope.DATA_WRITE) +@boundHandler +@autoDescribeRoute( + Description( + "Delete a terminal job and its owned output folder + staged inputs." + ) + .notes( + "A pending or running job returns 409 (cancel it first). A terminal job " + "is removed together with its private output folder and any remaining " + "staged inputs — deleting the job also deletes its results." + ) + .param("jobId", "The job identifier.", paramType="path") + .errorResponse("Write access was denied for the job.", 403) + .errorResponse("The job is still running.", 409) +) +def deleteJob(self, jobId): + user = self.getCurrentUser() + + model = girder_job.Job() + job = model.load(jobId, user=user, level=AccessType.WRITE, exc=True) + # The model.job.remove handler ALSO enforces this (protecting other + # Job.remove callers); the route returns the typed product response + # rather than the model's raise. + if not results.isTerminalStatus(job.get("status")): + raise RestException( + "Job is still running; cancel it before deleting", code=409 + ) + # The ownership cascade (owned output folder + staged inputs) lives in the + # model.job.remove handler; this route must not add a second cascade. + model.remove(job) + cherrypy.response.status = 204 + return None + + +@access.public(cookie=True, scope=TokenScope.DATA_READ) +@boundHandler +@autoDescribeRoute( + Description("Get job results.") + .param("jobId", "The job identifier.", paramType="path") + .produces(["application/json"]) +) +def getJobResults(self, jobId): + user = self.getCurrentUser() + + job = girder_job.Job().load(jobId, user=user, level=AccessType.READ, exc=True) + payload = results._jobResultsPayload(job, user) + if payload.get("code"): + cherrypy.response.status = 409 + if payload["resultState"] == "waiting": + cherrypy.response.headers["Retry-After"] = "2" + return payload + + +@access.public(cookie=True, scope=TokenScope.DATA_WRITE) +@boundHandler +@autoDescribeRoute( + Description("Cancel a job.") + .notes( + "Best-effort: Girder only transitions an INACTIVE/QUEUED/RUNNING job to " + "CANCELED, so cancelling an already-terminal job is a no-op. The response " + "is the job's real projected status after the attempt -- never a " + "fabricated 'cancelled' -- and the client's poller converges on whatever " + "terminal state Girder ultimately reports." + ) + .param("jobId", "The job identifier.", paramType="path") + .produces(["application/json"]) + .errorResponse("Write access was denied for the job.", 403) +) +def cancelJob(self, jobId): + user = self.getCurrentUser() + + jobModel = girder_job.Job() + job = jobModel.load(jobId, user=user, level=AccessType.WRITE, exc=True) + try: + jobModel.cancelJob(job) + except ValidationException: + # Girder refuses a CANCELED transition from a terminal state, so an + # already-finished job cannot be cancelled. Fall through and report the + # job's real state rather than fabricate a `cancelled` the poller would + # contradict. + pass + fresh = _loadJobForStatusProjection(jobId, user) + return results._projectJobStatus(fresh, user) + + +@access.public(cookie=True, scope=TokenScope.DATA_WRITE) +@boundHandler +@autoDescribeRoute( + Description("Stage a parent-bound labelmap as a transient processing input.") + .notes( + "Accepts multipart labelmap bytes plus a neutral reference-image InputValue. " + "The backend validates and resolves that opaque relationship against durable " + "reference files before minting the staged URI. The created item is tagged " + "transient, deleted when its job reaches a terminal state, or swept if " + "never submitted." + ) + .modelParam("folderId", model=Folder, level=AccessType.WRITE) + .param("file", "The labelmap bytes.", paramType="formData", dataType="file") + .jsonParam( + "descriptor", + "Typed labelmap resource descriptor.", + paramType="formData", + requireObject=True, + ) + .errorResponse() +) +def stageInput(self, folder, file, descriptor): + user = self.getCurrentUser() + # Validate the whole descriptor (reference image included) before writing + # bytes so a malformed, foreign, unauthorized, or transient reference never + # leaves an orphan upload. + name = inputs.validateStagedDescriptor(descriptor, user) + # Job-end cleanup never sees an upload that was never submitted, so age out + # this folder's orphans before adding another. + inputs._sweepOrphanTransients(folder) + fileDoc = inputs._streamMultipartFileIntoItem(folder, user, file, name) + try: + inputs._tagItemTransient(fileDoc) + except Exception: + # An untagged item is invisible to both the TTL sweep and the + # launch-manifest exclusion (each keys on the transient marker), so it + # would linger as apparent durable launch data. + inputs._removeTransientItems([fileDoc["itemId"]]) + raise + # The backend mints the staged URI; the client constructs none. + return {"uris": [makeFileDownloadUrl(fileDoc)]} + + +class _JobResource(Resource): + """Folder-free REST surface for the job-addressed routes. + + Mounted at ``/volview_processing`` (a sibling of ``/folder``, ``/item``), + this hosts status / results / cancel keyed by job id alone -- the launch + folder is not part of a job's identity. The launch-context routes + (tasks / spec / run / stage) stay on the folder tree because they operate + per-folder. The handlers are the same module-level ``@boundHandler`` + functions; only their mount point differs. + """ + + def __init__(self): + super().__init__() + self.resourceName = PROCESSING_ROUTE_NAME + self.route("GET", ("jobs", ":jobId"), getJob) + self.route("GET", ("jobs", ":jobId", "detail"), getJobHistoryDetail) + self.route("DELETE", ("jobs", ":jobId"), deleteJob) + self.route("GET", ("jobs", ":jobId", "results"), getJobResults) + self.route("POST", ("jobs", ":jobId", "cancel"), cancelJob) + + +def _ensureJobHistoryIndexesInBackground(): + """Kick the index builds off a daemon thread so plugin load never blocks. + + ``create_index`` is idempotent and a no-op on steady state, but the FIRST + boot against a large pre-existing jobs collection waits for the whole build; + the queries the indexes back merely degrade to scans until the build lands. + """ + + def build(): + try: + ensureJobHistoryIndexes() + except Exception: + logger.exception("Failed to ensure volview job-history indexes") + + threading.Thread( + target=build, name="volview-job-history-indexes", daemon=True + ).start() + + +def addBackendRoutes(info): + _ensureJobHistoryIndexesInBackground() + # Delete a job's transient staged inputs once it reaches a terminal state. + # Fires for every job update but no-ops cheaply unless the job carries the + # transient marker. + events.bind( + "jobs.job.update.after", + "girder_volview.backend.routes", + inputs._cleanupTransientOnJobDone, + ) + # Record each finalized output file's id onto the job that OWNS the file's + # private parent folder, keyed by output identifier, so result collection + # reads ids OFF the job. Fires for every upload but returns early unless the + # upload lands in a job's output folder under a declared identifier. + events.bind( + "model.file.finalizeUpload.after", + "girder_volview.backend.outputs", + outputs._recordJobOutput, + ) + # Ownership cascade: each job owns one private output folder + its staged + # inputs. Running before the DB delete, this refuses to remove a nonterminal + # owned job and cascade-deletes its owned resources, so the DELETE route, + # Girder's built-in job route, and any direct Job.remove caller all + # honor the same terminal guard and cleanup. + events.bind( + "model.job.remove", + "girder_volview.backend.outputs", + outputs._cascadeDeleteJobOwnedResources, + ) + # Reverse ownership cascade: deleting a job's output folder in the Girder + # hierarchy deletes the job record too (refusing for a live job), so folder + # deletion is a first-class "delete this job" gesture and no orphaned job + # rows accumulate. Removing the volview-jobs container recurses per job + # folder. + events.bind( + "model.folder.remove", + "girder_volview.backend.outputs", + outputs._cascadeDeleteFolderOwnedJob, + ) + # REST pre-guard for the same invariant: Folder.remove cleans contents + # BEFORE model.folder.remove fires, so refuse a live job's folder (or a + # container holding one) before the delete handler touches anything. + events.bind( + "rest.delete.folder/:id.before", + "girder_volview.backend.outputs", + outputs._refuseLiveJobFolderRestDelete, + ) + # The same preflight for every OTHER recursive deletion entry point: + # collection delete, user delete, and the batch /resource route all reach + # Folder.remove without passing DELETE /folder/:id, so each would otherwise + # destroy a live job's staged inputs and partial outputs before the late + # model-level guard could refuse. + events.bind( + "rest.delete.collection/:id.before", + "girder_volview.backend.outputs", + outputs._refuseLiveJobCollectionRestDelete, + ) + events.bind( + "rest.delete.user/:id.before", + "girder_volview.backend.outputs", + outputs._refuseLiveJobUserRestDelete, + ) + events.bind( + "rest.delete.resource.before", + "girder_volview.backend.outputs", + outputs._refuseLiveJobResourceRestDelete, + ) + # The recorded id map is READ-exposed; the job's own ACL is the gate + # (otherFields + exposeFields, mirroring slicer_cli_web's slicerCLIBindings). + + girder_job.Job().exposeFields( + level=AccessType.READ, fields={outputs._OUTPUTS_FIELD} + ) + info["apiRoot"].folder.route( + "GET", (":folderId", PROCESSING_ROUTE_NAME, "tasks"), listTasks + ) + # Job re-discovery is context-scoped (it takes a launch folder), not + # job-addressed: a reloaded client GETs this to re-find its jobs. + info["apiRoot"].folder.route( + "GET", (":folderId", PROCESSING_ROUTE_NAME, "jobs"), listJobHistory + ) + info["apiRoot"].folder.route( + "POST", (":folderId", PROCESSING_ROUTE_NAME, "stage"), stageInput + ) + info["apiRoot"].folder.route( + "GET", + (":folderId", PROCESSING_ROUTE_NAME, "tasks", ":taskId", "spec"), + getTaskSpec, + ) + info["apiRoot"].folder.route( + "POST", + (":folderId", PROCESSING_ROUTE_NAME, "tasks", ":taskId", "run"), + runTask, + ) + info["apiRoot"].volview_processing = _JobResource() diff --git a/girder_volview/backend/slicer_spec.py b/girder_volview/backend/slicer_spec.py new file mode 100644 index 0000000..267d31b --- /dev/null +++ b/girder_volview/backend/slicer_spec.py @@ -0,0 +1,816 @@ +"""Slicer Execution Model XML -> VolView task spec. + +The server emits VolView's ``zod``-defined task spec so the client never parses +a backend's XML. VolView's ``backend-contract`` ``task-spec`` golden fixtures +pin the output exactly. + +Pure standard library (``xml.etree``) so it imports without Girder. +""" + +import math +import re +import xml.etree.ElementTree as ET + + +def _first_child(el, tag): + return next((c for c in el if c.tag == tag), None) + + +def _all_children(el, tag): + return [c for c in el if c.tag == tag] + + +def _child_text(el, tag): + child = _first_child(el, tag) + if child is None or child.text is None: + return "" + return child.text + + +# Mapping tables. DO NOT redesign these: the fixtures pin them byte for byte. + +# Slicer element tag -> widget type. An unmapped tag yields ``None`` (the caller +# treats it as an unknown field kind, fail closed). +_TYPE_MAP = { + "integer": "number", + "float": "number", + "double": "number", + "boolean": "boolean", + "string": "string", + "integer-vector": "number-vector", + "float-vector": "number-vector", + "double-vector": "number-vector", + "string-vector": "string-vector", + "integer-enumeration": "number-enumeration", + "float-enumeration": "number-enumeration", + "double-enumeration": "number-enumeration", + "string-enumeration": "string-enumeration", + "region": "region", + "image": "image", + "file": "file", + "item": "item", + "directory": "directory", + "multi": "multi", +} + + +def _widget_type(tag): + return _TYPE_MAP.get(tag) + + +# Leading numeric run, JS ``parseFloat``-style (optional sign, digits, decimal, +# exponent). +_LEADING_FLOAT = re.compile(r"[+-]?(?:\d+\.?\d*|\.\d+)(?:[eE][+-]?\d+)?") + + +def _parse_float(value): + # Emulate JS ``parseFloat``: a lenient value (``"1,000"``, ``"50%"``, + # ``"1.5x"``) degrades to its leading number rather than raising + # ``ValueError`` and 500-ing the whole task-spec endpoint. No leading number + # yields NaN. + try: + return float(value) + except (TypeError, ValueError): + match = _LEADING_FLOAT.match(str(value).strip()) + return float(match.group(0)) if match else float("nan") + + +def _convert(widget_type, value): + """Coerce a raw XML string to the widget's value type.""" + if widget_type in ("number", "number-enumeration"): + return _parse_float(value) + if widget_type == "boolean": + return value.lower() == "true" + if widget_type == "number-vector": + return [_parse_float(s) for s in value.split(",")] + if widget_type == "string-vector": + return value.split(",") + return value + + +def _parse_constraints(widget_type, constraints_el): + """```` -> ``{min,max,step}`` (converted).""" + if constraints_el is None: + return {} + spec = {} + minimum = _child_text(constraints_el, "minimum") + maximum = _child_text(constraints_el, "maximum") + step = _child_text(constraints_el, "step") + if minimum: + spec["min"] = _convert(widget_type, minimum) + if maximum: + spec["max"] = _convert(widget_type, maximum) + if step: + spec["step"] = _convert(widget_type, step) + return spec + + +def _parse_default(widget_type, default_el): + """```` -> the converted value, or ``None``. + + Template placeholders (``{{x}}``) are skipped. + """ + if default_el is None: + return None + text = default_el.text or "" + if len(text) == 0: + return None + is_template = text[:2] == "{{" and text[-2:] == "}}" + if is_template: + return None + return _convert(widget_type, text) + + +def _parse_param(param_el, section): + tag = param_el.tag + widget = _widget_type(tag) + channel = "output" if _child_text(param_el, "channel") == "output" else "input" + # ctk_cli identifies a -less param by its longflag with leading dashes + # stripped; slicer_cli_web binds submitted args by that identifier, so the + # id must match or the submitted value is silently ignored. + param_id = ( + _child_text(param_el, "name") or _child_text(param_el, "longflag").lstrip("-") + ).strip() + required = len(_child_text(param_el, "index")) > 0 + values = None + if widget in ("string-enumeration", "number-enumeration"): + values = [ + _convert(widget, el.text or "") for el in _all_children(param_el, "element") + ] + return { + "tag": tag, # the raw Slicer element name + "widget": widget, # WidgetType, or None for an unmapped tag + "channel": channel, + "id": param_id, + "title": _child_text(param_el, "label"), + "help": _child_text(param_el, "description"), + "section": section, + "required": required, + "imageType": param_el.get("type"), # input type -> accepts + "fileExtensions": param_el.get("fileExtensions"), + "values": values, + "default": _parse_default(widget, _first_child(param_el, "default")), + "constraints": _parse_constraints( + widget, _first_child(param_el, "constraints") + ), + } + + +def _parse_panel(panel_el): + """Group a ```` panel's children by their leading ``