Asynchronous Download API endpoint

[IgorBalog-Eng]

Current implementation of API download endpoint is synchronous, meaning that if it is downloading large file, which takes several minutes to transfer, it will block response from controller.

We should improve this logic and return response OK, transfer started (with maybe some additional information) and unblock that servlet.
This will also require changes on UI, since it is dependent and relying on transfer to complete (response from controller) to handle logic for displaying buttons.

Proposition:

Phase 1:

• make changes in DataTransferAPIController.downloadData, to initiate transfer logic in another thread, or some other mechanism (publish event) and return response back, like suggested.
• create logic that will inform UI that transfer is completed. This will require to investigate possible solutions for this, how to pass such information to UI app.
• create logic in backend, that will provide information for specific transfer process status, something like 10%, 20%, 30% downloaded for specific transfer process (might require to calculate whole file size divided by chunk size) - this requirement is nice to have

Phase 2:
Once async download is completed, we should handle logic to terminate transfer, which will affect stopping current data transfer and perform cleanup of bucket (already imported parts)

• when request is received on CONSUMER side, we should stop inputStream processing and do bucket cleanup
• verify that TransferProcess status on both sides is TERMIANTED
• verify that bucket on CONSUMER side is cleaned from already uploaded parts

Investigate if it is possible to initiate TERMINATE from PROVIDER side, since COSNUMER is the one that handles data transfer (download) using presignedURL. Should be something like

• consumer receives terminate request on callbackController
• DataTransferService.terminateDataTransfer is handling this request, but it is also used on both roles, CONSUMER and PROVIDER, so we might distingush those 2 use-cases (if even applciable for PROVIDER)

[IgorBalog-Eng]

Starting point implementation for tracking transfers and logic for terminating and bucket cleanup in case of interruption/termination

TransferStatusTracker.txt

DataTransferAPIService

  public JsonNode terminateTransfer(String transferProcessId) {
        TransferProcess transferProcess = findTransferProcessById(transferProcessId);

        // First terminate any active transfer
        transferStatusTracker.terminateTransfer(transferProcessId);

... remaining logic

Consider refactoring DataTransferAPIService since there are a lot constructor parameters for initialization (if applicable to refactor)

HttpPullTransferStrategy.txt

[IgorBalog-Eng]

Improvement that includes using async calls in different way:

public interface DataTransferStrategy {
    CompletableFuture<Void> transfer(TransferProcess transferProcess);
}

strategy

@Service
@Slf4j
public class HttpPullTransferStrategy implements DataTransferStrategy {

    @Override
    public CompletableFuture<Void> transfer(TransferProcess transferProcess) {
        log.info("Executing HTTP PULL transfer for process {}", transferProcess.getId());

        String authorization = extractAuthorization(transferProcess);

        return downloadAndUploadToS3(
                transferProcess.getDataAddress().getEndpoint(),
                authorization,
                transferProcess.getId()
        ).thenAccept(key -> 
            log.info("Stored transfer process id - {} data!", key));
    }

    // Rest of the class remains the same
}

service

public CompletableFuture<Void> downloadData(String transferProcessId) {
    TransferProcess transferProcess = findTransferProcessById(transferProcessId);

    if (!transferProcess.getState().equals(TransferState.STARTED)) {
        log.error("Download aborted, Transfer Process is not in STARTED state");
        return CompletableFuture.failedFuture(
            new DataTransferAPIException("Download aborted, Transfer Process is not in STARTED state"));
    }

    policyCheck(transferProcess);

    log.info("Starting download transfer process id - {} data...", transferProcessId);

    // Get appropriate strategy and execute transfer
    DataTransferStrategy strategy = dataTransferStrategyFactory.getStrategy(transferProcess.getFormat());

    return strategy.transfer(transferProcess)
        .thenAccept(unused -> {
            TransferProcess transferProcessWithData = TransferProcess.Builder.newInstance()
                    .id(transferProcess.getId())
                    .agreementId(transferProcess.getAgreementId())
                    .consumerPid(transferProcess.getConsumerPid())
                    .providerPid(transferProcess.getProviderPid())
                    .callbackAddress(transferProcess.getCallbackAddress())
                    .dataAddress(transferProcess.getDataAddress())
                    .isDownloaded(true)
                    .dataId(transferProcessId)
                    .format(transferProcess.getFormat())
                    .state(transferProcess.getState())
                    .role(transferProcess.getRole())
                    .datasetId(transferProcess.getDatasetId())
                    .created(transferProcess.getCreated())
                    .createdBy(transferProcess.getCreatedBy())
                    .modified(transferProcess.getModified())
                    .lastModifiedBy(transferProcess.getLastModifiedBy())
                    .version(transferProcess.getVersion())
                    .build();

            transferProcessRepository.save(transferProcessWithData);
            log.info("Transfer process {} updated with downloaded status", transferProcessId);
        });
}

controller

@GetMapping(path = { "/{transferProcessId}/download" })
public ResponseEntity<GenericApiResponse<String>> downloadData(
        @PathVariable String transferProcessId) {
    log.info("Initiating async download for transfer process id - {}", transferProcessId);
    
    apiService.downloadData(transferProcessId)
        .whenComplete((result, throwable) -> {
            if (throwable != null) {
                log.error("Download failed for process {}: {}", 
                    transferProcessId, throwable.getMessage(), throwable);
            } else {
                log.info("Download completed successfully for process {}", transferProcessId);
            }
        });
    
    return ResponseEntity.accepted()
        .contentType(MediaType.APPLICATION_JSON)
        .body(GenericApiResponse.success(
            "Download started for transfer process " + transferProcessId,
            "Request accepted"));
}