Improved handling of Globus uploads (experimental async framework)

doc/release-notes/10623-globus-improvements.md

@@ -0,0 +1 @@
+A new alternative implementation of Globus polling during upload data transfers has been added in this release. This experimental framework does not rely on the instance staying up continuously for the duration of the transfer and saves the state information about Globus upload requests in the database. See the `globus-use-experimental-async-framework` feature flag in the Configuration guide.

doc/sphinx-guides/source/developers/big-data-support.rst

@@ -187,3 +187,5 @@ As described in that document, Globus transfers can be initiated by choosing the
 An overview of the control and data transfer interactions between components was presented at the 2022 Dataverse Community Meeting and can be viewed in the `Integrations and Tools Session Video <https://youtu.be/3ek7F_Dxcjk?t=5289>`_ around the 1 hr 28 min mark.
 
 See also :ref:`Globus settings <:GlobusSettings>`.
+
+An alternative, experimental implementation of Globus polling of ongoing upload transfers has been added in v6.4. This framework does not rely on the instance staying up continuously for the duration of the transfer and saves the state information about Globus upload requests in the database. Due to its experimental nature it is not enabled by default. See the ``globus-use-experimental-async-framework`` feature flag and the JVM option ``dataverse.globus.taskMonitoringServer`` described in the Configuration guide.

doc/sphinx-guides/source/developers/globus-api.rst

@@ -185,6 +185,8 @@ As the transfer can take significant time and the API call is asynchronous, the
 
 Once the transfer completes, Dataverse will remove the write permission for the principal.
 
+An alternative, experimental implementation of Globus polling of ongoing upload transfers has been added in v6.4. This new framework does not rely on the instance staying up continuously for the duration of the transfer and saves the state information about Globus upload requests in the database. Due to its experimental nature it is not enabled by default. See the ``globus-use-experimental-async-framework`` feature flag and the JVM option ``dataverse.globus.taskMonitoringServer`` described in the Configuration guide.
+
 Note that when using a managed endpoint that uses the Globus S3 Connector, the checksum should be correct as Dataverse can validate it. For file-based endpoints, the checksum should be included if available but Dataverse cannot verify it.
 
 In the remote/reference case, where there is no transfer to monitor, the standard /addFiles API call (see :ref:`direct-add-to-dataset-api`) is used instead. There are no changes for the Globus case.

doc/sphinx-guides/source/installation/config.rst

@@ -3259,6 +3259,13 @@ The email for your institution that you'd like to appear in bag-info.txt. See :r
 
 Can also be set via *MicroProfile Config API* sources, e.g. the environment variable ``DATAVERSE_BAGIT_SOURCEORG_EMAIL``.
 
+.. _dataverse.globus.taskMonitoringServer:
+
+dataverse.globus.taskMonitoringServer
++++++++++++++++++++++++++++++++++++++
+
+This setting is required in conjunction with the ``globus-use-experimental-async-framework`` feature flag. Setting it to true designates the Dataverse instance to serve as the dedicated polling server. It is needed so that the new framework can be used in a multi-node installation. 
+
 .. _feature-flags:
 
 Feature Flags
@@ -3294,6 +3301,9 @@ please find all known feature flags below. Any of these flags can be activated u
     * - disable-return-to-author-reason
       - Removes the reason field in the `Publish/Return To Author` dialog that was added as a required field in v6.2 and makes the reason an optional parameter in the :ref:`return-a-dataset` API call. 
       - ``Off``
+    * - globus-use-experimental-async-framework
+      - Activates a new experimental implementation of Globus polling of ongoing remote data transfers that does not rely on the instance staying up continuously for the duration of the transfers and saves the state information about Globus upload requests in the database. Added in v6.4. Note that the JVM option ``dataverse.globus.taskMonitoringServer`` described above must also be enabled on one (and only one, in a multi-node installation) Dataverse instance. 
+      - ``Off``
 
 
 **Note:** Feature flags can be set via any `supported MicroProfile Config API source`_, e.g. the environment variable

src/main/java/edu/harvard/iq/dataverse/DatasetServiceBean.java

@@ -412,12 +412,20 @@ public boolean checkDatasetLock(Long datasetId) {
         List<DatasetLock> lock = lockCounter.getResultList();
         return lock.size()>0;
     }
-
+
+    public List<DatasetLock> getLocksByDatasetId(Long datasetId) {
+        TypedQuery<DatasetLock> locksQuery = em.createNamedQuery("DatasetLock.getLocksByDatasetId", DatasetLock.class);
+        locksQuery.setParameter("datasetId", datasetId);
+        return locksQuery.getResultList();
+    }
+
     public List<DatasetLock> getDatasetLocksByUser( AuthenticatedUser user) {
 
         return listLocks(null, user);
     }
 
+    // @todo: we'll be better off getting rid of this method and using the other 
+    // version of addDatasetLock() (that uses datasetId instead of Dataset). 
     @TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
     public DatasetLock addDatasetLock(Dataset dataset, DatasetLock lock) {
         lock.setDataset(dataset);
@@ -467,6 +475,7 @@ public DatasetLock addDatasetLock(Long datasetId, DatasetLock.Reason reason, Lon
      * is {@code aReason}.
      * @param dataset the dataset whose locks (for {@code aReason}) will be removed.
      * @param aReason The reason of the locks that will be removed.
+     * @todo this should probably take dataset_id, not a dataset
      */
     @TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
     public void removeDatasetLocks(Dataset dataset, DatasetLock.Reason aReason) {

src/main/java/edu/harvard/iq/dataverse/EditDatafilesPage.java

@@ -2121,8 +2121,12 @@ public void handleFileUpload(FileUploadEvent event) throws IOException {
     }
 
     /**
-     * Using information from the DropBox choose, ingest the chosen files
-     * https://www.dropbox.com/developers/dropins/chooser/js
+     * External, aka "Direct" Upload. 
+     * The file(s) have been uploaded to physical storage (such as S3) directly,
+     * this call is to create and add the DataFiles to the Dataset on the Dataverse 
+     * side. The method does NOT finalize saving the datafiles in the database -
+     * that will happen when the user clicks 'Save', similar to how the "normal"
+     * uploads are handled. 
      *
      * @param event
      */

src/main/java/edu/harvard/iq/dataverse/ExternalFileUploadInProgress.java

@@ -0,0 +1,114 @@
+/*
+ * Click nbfs://nbhost/SystemFileSystem/Templates/Licenses/license-default.txt to change this license
+ * Click nbfs://nbhost/SystemFileSystem/Templates/Classes/Class.java to edit this template
+ */
+package edu.harvard.iq.dataverse;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Index;
+import jakarta.persistence.NamedQueries;
+import jakarta.persistence.NamedQuery;
+import jakarta.persistence.Table;
+import java.io.Serializable;
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+
+/**
+ *
+ * @author landreev
+ * 
+ * The name of the class is provisional. I'm open to better-sounding alternatives,
+ * if anyone can think of any. 
+ * But I wanted to avoid having the word "Globus" in the entity name. I'm adding 
+ * it specifically for the Globus use case. But I'm guessing there's a chance 
+ * this setup may come in handy for other types of datafile uploads that happen
+ * externally. (?)
+ */
+@NamedQueries({
+	@NamedQuery( name="ExternalFileUploadInProgress.deleteByTaskId",
+		query="DELETE FROM ExternalFileUploadInProgress f WHERE f.taskId=:taskId"),
+        @NamedQuery(name = "ExternalFileUploadInProgress.findByTaskId",
+                query = "SELECT f FROM ExternalFileUploadInProgress f WHERE f.taskId=:taskId")})
+@Entity
+@Table(indexes = {@Index(columnList="taskid")})
+public class ExternalFileUploadInProgress implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    public Long getId() {
+        return id;
+    }
+
+    public void setId(Long id) {
+        this.id = id;
+    }
+
+    /**
+     * Rather than saving various individual fields defining the datafile, 
+     * which would essentially replicate the DataFile table, we are simply 
+     * storing the full json record as passed to the API here.  
+     */
+    @Column(columnDefinition = "TEXT", nullable=false)
+    private String fileInfo;
+
+    /**
+     * This is Globus-specific task id associated with the upload in progress
+     */  
+    @Column(nullable=false)
+    private String taskId; 
+
+    public ExternalFileUploadInProgress() {
+    }
+
+    public ExternalFileUploadInProgress(String taskId, String fileInfo) {
+        this.taskId = taskId;
+        this.fileInfo = fileInfo;
+    }
+
+    public String getFileInfo() {
+        return fileInfo;
+    }
+
+    public void setFileInfo(String fileInfo) {
+        this.fileInfo = fileInfo;
+    }
+
+    public String getTaskId() {
+        return taskId;
+    }
+
+    public void setTaskId(String taskId) {
+        this.taskId = taskId;
+    }
+
+    @Override
+    public int hashCode() {
+        int hash = 0;
+        hash += (id != null ? id.hashCode() : 0);
+        return hash;
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        // TODO: Warning - this method won't work in the case the id fields are not set
+        if (!(object instanceof ExternalFileUploadInProgress)) {
+            return false;
+        }
+        ExternalFileUploadInProgress other = (ExternalFileUploadInProgress) object;
+        if ((this.id == null && other.id != null) || (this.id != null && !this.id.equals(other.id))) {
+            return false;
+        }
+        return true;
+    }
+
+    @Override
+    public String toString() {
+        return "edu.harvard.iq.dataverse.ExternalFileUploadInProgress[ id=" + id + " ]";
+    }
+
+}

src/main/java/edu/harvard/iq/dataverse/MailServiceBean.java

@@ -623,6 +623,7 @@ public String getMessageTextBasedOnNotification(UserNotification userNotificatio
                         comment
                 ))  ;
                 return downloadCompletedMessage;
+
             case GLOBUSUPLOADCOMPLETEDWITHERRORS:
                 dataset =  (Dataset) targetObject;
                 messageText = BundleUtil.getStringFromBundle("notification.email.greeting.html");
@@ -633,8 +634,30 @@ public String getMessageTextBasedOnNotification(UserNotification userNotificatio
                         comment
                 ))  ;
                 return  uploadCompletedWithErrorsMessage;
+
+            case GLOBUSUPLOADREMOTEFAILURE:
+                dataset =  (Dataset) targetObject;
+                messageText = BundleUtil.getStringFromBundle("notification.email.greeting.html");
+                String uploadFailedRemotelyMessage = messageText + BundleUtil.getStringFromBundle("notification.mail.globus.upload.failedRemotely", Arrays.asList(
+                        systemConfig.getDataverseSiteUrl(),
+                        dataset.getGlobalId().asString(),
+                        dataset.getDisplayName(),
+                        comment
+                ))  ;
+                return  uploadFailedRemotelyMessage;
 
-            case GLOBUSDOWNLOADCOMPLETEDWITHERRORS:
+            case GLOBUSUPLOADLOCALFAILURE:
+                dataset =  (Dataset) targetObject;
+                messageText = BundleUtil.getStringFromBundle("notification.email.greeting.html");
+                String uploadFailedLocallyMessage = messageText + BundleUtil.getStringFromBundle("notification.mail.globus.upload.failedLocally", Arrays.asList(
+                        systemConfig.getDataverseSiteUrl(),
+                        dataset.getGlobalId().asString(),
+                        dataset.getDisplayName(),
+                        comment
+                ))  ;
+                return  uploadFailedLocallyMessage;
+
+                case GLOBUSDOWNLOADCOMPLETEDWITHERRORS:
                 dataset =  (Dataset) targetObject;
                 messageText = BundleUtil.getStringFromBundle("notification.email.greeting.html");
                 String downloadCompletedWithErrorsMessage = messageText + BundleUtil.getStringFromBundle("notification.mail.globus.download.completedWithErrors", Arrays.asList(
@@ -763,6 +786,8 @@ public Object getObjectOfNotification (UserNotification userNotification){
                 return versionService.find(userNotification.getObjectId());
             case GLOBUSUPLOADCOMPLETED:
             case GLOBUSUPLOADCOMPLETEDWITHERRORS:
+            case GLOBUSUPLOADREMOTEFAILURE:
+            case GLOBUSUPLOADLOCALFAILURE: 
             case GLOBUSDOWNLOADCOMPLETED:
             case GLOBUSDOWNLOADCOMPLETEDWITHERRORS:
                 return datasetService.find(userNotification.getObjectId());

src/main/java/edu/harvard/iq/dataverse/UserNotification.java

@@ -39,7 +39,8 @@ public enum Type {
         CHECKSUMIMPORT, CHECKSUMFAIL, CONFIRMEMAIL, APIGENERATED, INGESTCOMPLETED, INGESTCOMPLETEDWITHERRORS, 
         PUBLISHFAILED_PIDREG, WORKFLOW_SUCCESS, WORKFLOW_FAILURE, STATUSUPDATED, DATASETCREATED, DATASETMENTIONED,
         GLOBUSUPLOADCOMPLETED, GLOBUSUPLOADCOMPLETEDWITHERRORS,
-        GLOBUSDOWNLOADCOMPLETED, GLOBUSDOWNLOADCOMPLETEDWITHERRORS, REQUESTEDFILEACCESS;
+        GLOBUSDOWNLOADCOMPLETED, GLOBUSDOWNLOADCOMPLETEDWITHERRORS, REQUESTEDFILEACCESS, 
+        GLOBUSUPLOADREMOTEFAILURE, GLOBUSUPLOADLOCALFAILURE;
 
         public String getDescription() {
             return BundleUtil.getStringFromBundle("notification.typeDescription." + this.name());

src/main/java/edu/harvard/iq/dataverse/api/ApiConstants.java

@@ -17,4 +17,8 @@ private ApiConstants() {
     public static final String DS_VERSION_LATEST = ":latest";
     public static final String DS_VERSION_DRAFT = ":draft";
     public static final String DS_VERSION_LATEST_PUBLISHED = ":latest-published";
+
+    // addFiles call
+    public static final String API_ADD_FILES_COUNT_PROCESSED = "Total number of files";
+    public static final String API_ADD_FILES_COUNT_SUCCESSFULL = "Number of files successfully added";
 }

src/main/java/edu/harvard/iq/dataverse/api/Datasets.java

@@ -4009,6 +4009,7 @@ public Response addGlobusFilesToDataset(@Context ContainerRequestContext crc,
         logger.info(" ====  (api addGlobusFilesToDataset) jsonData   ====== " + jsonData);
 
         if (!systemConfig.isHTTPUpload()) {
+            // @todo why isHTTPUpload()? - shouldn't it be checking isGlobusUpload() here? 
             return error(Response.Status.SERVICE_UNAVAILABLE, BundleUtil.getStringFromBundle("file.api.httpDisabled"));
         }
 
@@ -4034,6 +4035,8 @@ public Response addGlobusFilesToDataset(@Context ContainerRequestContext crc,
             return wr.getResponse();
         }
 
+        // @todo check if the dataset is already locked!
+
         JsonObject jsonObject = null;
         try {
             jsonObject = JsonUtil.getJsonObject(jsonData);
@@ -4075,7 +4078,11 @@ public Response addGlobusFilesToDataset(@Context ContainerRequestContext crc,
         String requestUrl = SystemConfig.getDataverseSiteUrlStatic();
 
         // Async Call
-        globusService.globusUpload(jsonObject, token, dataset, requestUrl, authUser);
+        try {
+            globusService.globusUpload(jsonObject, token, dataset, requestUrl, authUser);
+        } catch (IllegalArgumentException ex) {
+            return badRequest("Invalid parameters: "+ex.getMessage());
+        }
 
         return ok("Async call to Globus Upload started ");
 

src/main/java/edu/harvard/iq/dataverse/authorization/providers/builtin/DataverseUserPage.java

@@ -528,6 +528,8 @@ public void displayNotification() {
                 case GLOBUSUPLOADCOMPLETEDWITHERRORS:
                 case GLOBUSDOWNLOADCOMPLETED:
                 case GLOBUSDOWNLOADCOMPLETEDWITHERRORS:
+                case GLOBUSUPLOADREMOTEFAILURE:
+                case GLOBUSUPLOADLOCALFAILURE: 
                     userNotification.setTheObject(datasetService.find(userNotification.getObjectId()));
                     break;
 

src/main/java/edu/harvard/iq/dataverse/datasetutility/AddReplaceFileHelper.java

@@ -2139,9 +2139,9 @@ public Response addFiles(String jsonData, Dataset dataset, User authUser) {
                     logger.log(Level.WARNING, "Dataset not locked for EditInProgress ");
                 } else {
                     datasetService.removeDatasetLocks(dataset, DatasetLock.Reason.EditInProgress);
-                    logger.log(Level.INFO, "Removed EditInProgress lock ");
+                    logger.log(Level.INFO, "Removed EditInProgress lock "+eipLock.getId());
                 }
-
+                
                 try {
                     Command<Dataset> cmd = new UpdateDatasetVersionCommand(dataset, dvRequest, clone);
                     ((UpdateDatasetVersionCommand) cmd).setValidateLenient(true);
@@ -2167,8 +2167,8 @@ public Response addFiles(String jsonData, Dataset dataset, User authUser) {
         }
 
         JsonObjectBuilder result = Json.createObjectBuilder()
-                .add("Total number of files", totalNumberofFiles)
-                .add("Number of files successfully added", successNumberofFiles);
+                .add(ApiConstants.API_ADD_FILES_COUNT_PROCESSED, totalNumberofFiles)
+                .add(ApiConstants.API_ADD_FILES_COUNT_SUCCESSFULL, successNumberofFiles);
 
 
         return Response.ok().entity(Json.createObjectBuilder()

src/main/java/edu/harvard/iq/dataverse/engine/command/impl/UpdateDatasetVersionCommand.java

@@ -14,6 +14,7 @@
 import edu.harvard.iq.dataverse.engine.command.RequiredPermissions;
 import edu.harvard.iq.dataverse.engine.command.exception.CommandException;
 import edu.harvard.iq.dataverse.engine.command.exception.IllegalCommandException;
+import edu.harvard.iq.dataverse.util.BundleUtil;
 import edu.harvard.iq.dataverse.util.DatasetFieldUtil;
 import edu.harvard.iq.dataverse.util.FileMetadataUtil;
 
@@ -102,7 +103,10 @@ public Dataset execute(CommandContext ctxt) throws CommandException {
         }
 
         Dataset theDataset = getDataset();        
-        ctxt.permissions().checkUpdateDatasetVersionLock(theDataset, getRequest(), this);
+        //ctxt.permissions().checkUpdateDatasetVersionLock(theDataset, getRequest(), this);
+        // this is an experiment (probably temporary)
+        checkUpdateDatasetVersionLock(ctxt);
+
         Dataset savedDataset = null;
 
         DatasetVersion persistedVersion = clone;
@@ -297,5 +301,23 @@ public boolean onSuccess(CommandContext ctxt, Object r) {
         ctxt.index().asyncIndexDataset((Dataset) r, true);
         return true;
     }
-
+
+    private void checkUpdateDatasetVersionLock(CommandContext ctxt) throws IllegalCommandException {
+        List<DatasetLock> locks = ctxt.datasets().getLocksByDatasetId(getDataset().getId());
+        //locks.forEach(lock -> {
+        for (DatasetLock lock : locks) {
+            // Ingest lock is ok: 
+            if (DatasetLock.Reason.Ingest != lock.getReason()) {
+                // with Workflow lock *some* users can edit;
+                // any other kind of lock - nope
+                if (DatasetLock.Reason.Workflow != lock.getReason()
+                        || !ctxt.permissions().isMatchingWorkflowLock(getDataset(),
+                                getUser().getIdentifier(),
+                                getRequest().getWFInvocationId())) {
+                    throw new IllegalCommandException(
+                            BundleUtil.getStringFromBundle("dataset.message.locked.editNotAllowed"), this);
+                }
+            }
+        }
+    }
 }