Add ndi.cloud.api.files.waitForFileUpload; wait for upload processing in uploadSingleFile and FilesTest

src/ndi/+ndi/+cloud/+api/+files/waitForFileUpload.m

@@ -0,0 +1,69 @@
+function [b, answer, apiResponse, apiURL] = waitForFileUpload(cloudDatasetID, cloudFileUIDs, options)
+%WAITFORFILEUPLOAD Wait until single-file uploads are recorded by the server.
+%
+%   [B, ANSWER, APIRESPONSE, APIURL] = ndi.cloud.api.files.waitForFileUpload(CLOUDDATASETID, CLOUDFILEUIDS)
+%   [...] = ndi.cloud.api.files.waitForFileUpload(CLOUDDATASETID, CLOUDFILEUIDS, 'timeout', T, ...)
+%
+%   A PUT to a signed upload URL returning 200 only means S3 has the object.
+%   The server then records the upload (uploaded=true, size, crc32) in an
+%   S3-triggered step that can lag the PUT by seconds or more when other
+%   uploads are queued on the same stage. This function polls
+%   ndi.cloud.api.files.lookupFiles at exponentially growing intervals until
+%   every uid is marked uploaded or the overall timeout elapses.
+%
+%   This is the single-file counterpart to waitForBulkUpload; bulk (zip)
+%   uploads should use that function, or putFiles(..., 'waitForCompletion',
+%   true), instead.
+%
+%   Inputs:
+%       cloudDatasetID - The cloud dataset id.
+%       cloudFileUIDs  - String array (or cellstr) of file uids to wait for.
+%
+%   Name-Value Pairs:
+%       'timeout'         (double) - Overall deadline in seconds. Default 120.
+%       'initialInterval' (double) - First sleep between polls (s). Default 1.
+%       'maxInterval'     (double) - Cap on the per-poll sleep (s). Default 10.
+%       'backoffFactor'   (double) - Multiplier applied after each poll. Default 2.
+%
+%   Outputs:
+%       b            - True iff every uid was marked uploaded before the timeout.
+%       answer       - Struct describing the final state:
+%                        .state   - 'complete' or 'timeout'
+%                        .files   - lookupFiles rows (uid, uploaded, size,
+%                                   crc32) for the uploaded uids, plus on
+%                                   timeout the last rows for pending ones
+%                        .pending - string array of uids not yet uploaded
+%                                   (including uids with no record)
+%                        .elapsed - seconds spent waiting
+%       apiResponse  - The ResponseMessage from the last poll.
+%       apiURL       - The URL of the last poll.
+%
+%   Example:
+%       [s, url] = ndi.cloud.api.files.getFileUploadURL('d-123', 'f-abc');
+%       ndi.cloud.api.files.putFiles(url, 'localfile.dat');
+%       b = ndi.cloud.api.files.waitForFileUpload('d-123', 'f-abc');
+%
+%   See also: ndi.cloud.api.implementation.files.WaitForFileUpload,
+%             ndi.cloud.api.files.lookupFiles,
+%             ndi.cloud.api.files.putFiles,
+%             ndi.cloud.api.files.waitForBulkUpload
+
+    arguments
+        cloudDatasetID (1,1) string
+        cloudFileUIDs (1,:) string
+        options.timeout (1,1) double {mustBePositive} = 120
+        options.initialInterval (1,1) double {mustBePositive} = 1
+        options.maxInterval (1,1) double {mustBePositive} = 10
+        options.backoffFactor (1,1) double {mustBePositive} = 2
+    end
+
+    api_call = ndi.cloud.api.implementation.files.WaitForFileUpload(...
+        'cloudDatasetID', cloudDatasetID, ...
+        'cloudFileUIDs', cloudFileUIDs, ...
+        'timeout', options.timeout, ...
+        'initialInterval', options.initialInterval, ...
+        'maxInterval', options.maxInterval, ...
+        'backoffFactor', options.backoffFactor);
+
+    [b, answer, apiResponse, apiURL] = api_call.execute();
+end

src/ndi/+ndi/+cloud/+api/+implementation/+files/WaitForFileUpload.m

@@ -0,0 +1,109 @@
+classdef WaitForFileUpload < ndi.cloud.api.call
+%WAITFORFILEUPLOAD Wait until single-file uploads are recorded by the server.
+%
+%   Implementation class behind ndi.cloud.api.files.waitForFileUpload.
+%   Polls lookupFiles(datasetId, uids) with exponential backoff until every
+%   uid is marked uploaded or the overall timeout elapses.
+%
+%   A signed PUT returning 200 only means S3 has the object; the server
+%   records the upload in an S3-triggered step afterwards. Use this between
+%   a single-file putFiles and anything that reads the file's record
+%   (listFiles, getFileDetails, a download URL).
+
+    properties
+        cloudFileUIDs (1,:) string
+        timeout (1,1) double = 120
+        initialInterval (1,1) double = 1
+        maxInterval (1,1) double = 10
+        backoffFactor (1,1) double = 2
+    end
+
+    methods
+        function this = WaitForFileUpload(args)
+            %WAITFORFILEUPLOAD Construct a new WaitForFileUpload call.
+            arguments
+                args.cloudDatasetID (1,1) string
+                args.cloudFileUIDs (1,:) string
+                args.timeout (1,1) double {mustBePositive} = 120
+                args.initialInterval (1,1) double {mustBePositive} = 1
+                args.maxInterval (1,1) double {mustBePositive} = 10
+                args.backoffFactor (1,1) double {mustBePositive} = 2
+            end
+            this.cloudDatasetID = args.cloudDatasetID;
+            this.cloudFileUIDs = unique(args.cloudFileUIDs, 'stable');
+            this.timeout = args.timeout;
+            this.initialInterval = args.initialInterval;
+            this.maxInterval = args.maxInterval;
+            this.backoffFactor = args.backoffFactor;
+            this.endpointName = 'lookup_dataset_files';
+        end
+
+        function [b, answer, apiResponse, apiURL] = execute(this)
+            %EXECUTE Poll until every uid is uploaded or the timeout elapses.
+            %
+            %   b           - true iff every uid was marked uploaded before
+            %                 the timeout. false on timeout; a failing
+            %                 lookupFiles call is retried until then.
+            %   answer      - struct with state ('complete' or 'timeout'),
+            %                 files (the uploaded rows, plus on timeout the
+            %                 last poll's rows for uids still pending),
+            %                 pending (uids not yet uploaded) and elapsed.
+            %   apiResponse - The ResponseMessage from the last poll.
+            %   apiURL      - The URL of the last poll.
+
+            b = false;
+            apiResponse = [];
+            apiURL = [];
+            doneFiles = []; % lookupFiles rows already marked uploaded
+            pendingFiles = []; % latest rows for uids not yet uploaded
+            pending = this.cloudFileUIDs;
+
+            deadline = tic;
+            interval = this.initialInterval;
+
+            while ~isempty(pending)
+                [ok, result, apiResponse, apiURL] = this.pollStatus(pending);
+
+                if ok && isstruct(result) && isfield(result, 'files')
+                    pendingFiles = [];
+                    if ~isempty(result.files)
+                        result.files = result.files(:);
+                        isUp = [result.files.uploaded];
+                        doneFiles = [doneFiles; result.files(isUp)]; %#ok<AGROW>
+                        pendingFiles = result.files(~isUp);
+                        pending = setdiff(pending, string({result.files(isUp).uid}), 'stable');
+                    end
+                    if isempty(pending)
+                        break;
+                    end
+                end
+
+                elapsed = toc(deadline);
+                if elapsed >= this.timeout
+                    answer = struct('state', 'timeout', 'files', [doneFiles; pendingFiles], ...
+                                    'pending', pending, 'elapsed', elapsed);
+                    return;
+                end
+
+                remaining = this.timeout - elapsed;
+                pause(min([interval, this.maxInterval, remaining]));
+                interval = min(interval * this.backoffFactor, this.maxInterval);
+            end
+
+            b = true;
+            answer = struct('state', 'complete', 'files', doneFiles, ...
+                            'pending', strings(1, 0), 'elapsed', toc(deadline));
+        end
+    end
+
+    methods (Access = protected)
+        function [ok, result, apiResponse, apiURL] = pollStatus(this, uids)
+            %POLLSTATUS Look up the given uids once. Seam: a test subclass
+            %   overrides this to script a sequence of lookupFiles results,
+            %   so the completion, timeout and retry logic can be exercised
+            %   without a server.
+            [ok, result, apiResponse, apiURL] = ...
+                ndi.cloud.api.files.lookupFiles(this.cloudDatasetID, uids);
+        end
+    end
+end

src/ndi/+ndi/+cloud/+sync/+internal/uploadFilesForDatasetDocuments.m

@@ -71,19 +71,35 @@
             app = ndi.gui.component.ProgressBarWindow('NDI tasks');
             uuid = did.ido.unique_id();
             app.addBar('Label','Uploading document-associated binary files','tag',uuid,'Auto',true);
+            sentUids = strings(1, 0);
             for i=1:numel(file_manifest)
                 if file_manifest(i).is_uploaded==false
+                    % Wait for the server to record the uploads once, below,
+                    % rather than after every file.
                     [upload_success, upload_message] = ndi.cloud.uploadSingleFile(cloudDatasetId, ...
-                        file_manifest(i).uid, file_manifest(i).file_path);
+                        file_manifest(i).uid, file_manifest(i).file_path, 'waitForUpload', false);
                     if ~upload_success
                         if success
                             message = ['Failed to upload file ' file_manifest(i).uid ': ' upload_message];
                         end
                         success = false;
+                    else
+                        sentUids(end+1) = string(file_manifest(i).uid); %#ok<AGROW>
                     end
                 end
                 app.updateBar(uuid,i/numel(file_manifest));
             end
+            if ~isempty(sentUids)
+                [wait_success, wait_answer] = ndi.cloud.api.files.waitForFileUpload(cloudDatasetId, sentUids);
+                if ~wait_success
+                    if success
+                        message = sprintf(['%d of %d uploaded files were not recorded by NDI Cloud ' ...
+                            'within %.0f s (first: %s).'], numel(wait_answer.pending), numel(sentUids), ...
+                            wait_answer.elapsed, wait_answer.pending(1));
+                    end
+                    success = false;
+                end
+            end
         case "batch"
             [batch_success, batch_message] = ndi.cloud.upload.zipForUpload(ndiDataset, file_manifest, totalSizeKb, cloudDatasetId);
             if ~batch_success

src/ndi/+ndi/+cloud/uploadSingleFile.m

@@ -19,6 +19,13 @@
 %                               corrupted upload and NDI Cloud records the
 %                               file's crc32. Ignored for bulk uploads, whose
 %                               zip entries carry their own CRC-32.
+%   'waitForUpload' (logical) - If true (default), return only once NDI Cloud
+%                               has recorded the upload (ndi.cloud.api.files.
+%                               waitForFileUpload; for bulk uploads, the zip
+%                               extraction job). A PUT returning 200 only
+%                               means S3 has the object; the server records
+%                               it shortly after. Set false when uploading
+%                               many files and wait for them all at once.
 %
 %   Outputs:
 %   B - true if the upload was successful, false otherwise.
@@ -31,6 +38,7 @@
         options.useBulkUpload (1,1) logical = false
         options.useCurl (1,1) logical = true
         options.checksum (1,1) logical = true
+        options.waitForUpload (1,1) logical = true
     end
 
     errormsg = '';
@@ -49,9 +57,20 @@
 
             [b_put, put_or_error] = ndi.cloud.api.files.putFiles(info_or_error.url, zip_file, ...
                 'useCurl', options.useCurl, ...
-                'jobId', info_or_error.jobId);
+                'jobId', info_or_error.jobId, ...
+                'waitForCompletion', options.waitForUpload);
             if ~b_put
-                error(['Could not upload zip file: ' put_or_error.message]);
+                if isstruct(put_or_error) && isfield(put_or_error, 'message')
+                    detail = put_or_error.message;
+                elseif isstruct(put_or_error) && isfield(put_or_error, 'state')
+                    % waitForCompletion's final bulk-upload status
+                    detail = ['bulk upload job ended in state ' char(string(put_or_error.state))];
+                elseif ischar(put_or_error) || isstring(put_or_error)
+                    detail = char(put_or_error);
+                else
+                    detail = jsonencode(put_or_error);
+                end
+                error(['Could not upload zip file: ' detail]);
             end
 
         catch e
@@ -85,6 +104,16 @@
             errormsg = ['Could not upload file: ' put_or_error.message];
             return;
         end
+
+        if options.waitForUpload
+            [b_wait, wait_answer] = ndi.cloud.api.files.waitForFileUpload(cloudDatasetID, cloudFileUID);
+            if ~b_wait
+                b = false;
+                errormsg = sprintf(['File was sent but NDI Cloud did not record the upload ' ...
+                    'within %.0f s (state: %s).'], wait_answer.elapsed, wait_answer.state);
+                return;
+            end
+        end
     end
 
     b = true;

tests/+ndi/+test/+helper/ScriptedFileUpload.m

@@ -0,0 +1,47 @@
+classdef ScriptedFileUpload < ndi.cloud.api.implementation.files.WaitForFileUpload
+% SCRIPTEDFILEUPLOAD - a WaitForFileUpload that returns canned lookups.
+%
+% Overrides the pollStatus seam so the completion, timeout and retry logic
+% can be exercised without a server. RESULTS is a cell array of structs
+% with fields:
+%   ok     - logical, what pollStatus should report
+%   result - the lookupFiles answer, normally a struct with 'files' (Nx1
+%            struct array: uid, uploaded, size, crc32) and 'missing'
+%
+% Calls past the end repeat the last entry, which is what a timeout test
+% wants. Give the constructor small intervals so the real pause() costs
+% milliseconds. polledUids records the uids asked for on each call.
+
+    properties
+        results = {}
+        callCount = 0
+        polledUids = {}
+    end
+
+    methods
+        function this = ScriptedFileUpload(results, varargin)
+            this@ndi.cloud.api.implementation.files.WaitForFileUpload(varargin{:});
+            this.results = results;
+        end
+    end
+
+    methods (Static)
+        function row = fileRow(uid, uploaded)
+            % One lookupFiles row, as lookupFilesResult returns it.
+            row = struct('uid', char(uid), 'uploaded', logical(uploaded), ...
+                'size', double(uploaded) * 47, 'crc32', '');
+        end
+    end
+
+    methods (Access = protected)
+        function [ok, result, apiResponse, apiURL] = pollStatus(this, uids)
+            this.callCount = this.callCount + 1;
+            this.polledUids{end+1} = uids;
+            entry = this.results{min(this.callCount, numel(this.results))};
+            ok = entry.ok;
+            result = entry.result;
+            apiResponse = [];
+            apiURL = "scripted://lookup";
+        end
+    end
+end