#434

backend/app/services/counts_export_jobs.rb

@@ -45,6 +45,20 @@ class CountsExportJobs
     write(state)
   end
 
+  def self.failure_code(error)
+    case error
+    when Cancelled then 'cancelled'
+    when CountsAxes::TooManyAxes then 'too_many_axes'
+    when CountsAxes::PrivateCondition then 'private_condition'
+    when CountsTableQuery::TooManyColumns then 'too_many_columns'
+    when ActiveRecord::StatementTimeout then 'counts_query_timeout'
+    when CountsExportDownload::TimedOut then 'generation_timeout'
+    when CountsExportDownload::TooLarge then 'export_too_large'
+    when CountsExportDownload::StorageFull, Errno::ENOSPC then 'export_storage_full'
+    else 'generation_failed'
+    end
+  end
+
   def self.directory = CountsExportDownload.directory.join('jobs')
   def self.verifier = Rails.application.message_verifier('counts_export_job')
 
@@ -176,6 +190,12 @@ class CountsExportJobs
       return unless state && state['owner'] == owner && state['expires_at'] > Time.current.to_f
 
       expire_stalled(state)
+      if (state['status'] == 'ready' &&
+          ['pending', 'streaming'].include?(state['download_status']) &&
+          state.fetch('download_expires_at', 0) <= Time.current.to_f)
+        state['download_status'] = 'timeout'
+        write(state)
+      end
 
       if cancel && ['queued', 'running', 'ready'].include?(state['status'])
         state['status'] = state['status'] == 'running' ? 'cancelling' : 'cancelled'
@@ -183,8 +203,11 @@ class CountsExportJobs
         state.delete('token')
         write(state)
       end
-      state.slice('status', 'phase', 'processed', 'total', 'format', 'token', 'error',
+      result = state.slice('status', 'phase', 'processed', 'total', 'format', 'token', 'error',
           'updated_at', 'heartbeat_at', 'download_attempt', 'download_status')
+      result['max_axes'] = CountsAxes::MAX_AXES if state['error'] == 'too_many_axes'
+      result['max_columns'] = CountsTableQuery::MAX_COLUMNS if state['error'] == 'too_many_columns'
+      result
     end
   end
 
@@ -198,18 +221,53 @@ class CountsExportJobs
           state['expires_at'] > Time.current.to_f && HistoryAccess.authorised_artifact?(state))
       if attempt
         return unless state['download_attempt'] == attempt && state['token'] == file_token
-        return unless ['pending', 'started'].include?(state['download_status'])
+        return unless state['download_status'] == 'started' ||
+            state.fetch('download_expires_at', 0) > Time.current.to_f
 
-        state['download_status'] = status if status
+        transitions = { 'streaming' => ['pending'], 'started' => ['streaming'],
+                        'failed' => ['pending', 'streaming', 'started'],
+                        'unavailable' => ['pending', 'streaming'] }
+        if status == 'active'
+          return state.slice('download_attempt', 'download_status', 'token') if
+              ['streaming', 'started'].include?(state['download_status'])
+          return
+        end
+        if status
+          return unless transitions.fetch(status, []).include?(state['download_status'])
+
+          state['download_status'] = status
+        else
+          return unless state['download_status'] == 'pending'
+        end
       else
+        # Repeated acceptance POSTs do not invalidate an in-flight GET or issue duplicates.
+        if (state['download_status'] == 'started' ||
+            (['pending', 'streaming'].include?(state['download_status']) &&
+             state.fetch('download_expires_at', 0) > Time.current.to_f))
+          return state.slice('download_attempt', 'download_status', 'token')
+        end
         state['download_attempt'] = SecureRandom.hex(24)
         state['download_status'] = 'pending'
+        state['download_expires_at'] = Time.current.to_f + 45
       end
       write(state)
       state.slice('download_attempt', 'download_status', 'token')
     end
   end
 
+  # GET already verified authority. During streaming, observe cancellation without
+  # repeating the six-table policy revision aggregate for every 64KiB chunk.
+  def self.stream_active?(token, owner, file_token:, attempt:)
+    id = authorised_id(token, owner)
+    state = id && read(id)
+    state && state['owner'] == owner && state['status'] == 'ready' &&
+        state['expires_at'] > Time.current.to_f && state['token'] == file_token &&
+        state['download_attempt'] == attempt &&
+        (state['download_status'] == 'started' ||
+         (state['download_status'] == 'streaming' &&
+          state.fetch('download_expires_at', 0) > Time.current.to_f))
+  end
+
   def self.checkpoint(id, processed = nil, total = nil, phase: nil)
     synchronise do
       state = read(id)
@@ -398,7 +456,7 @@ class CountsExportJobs
         if current
           unless current['status'] == 'failed'
             current['status'] = error.is_a?(Cancelled) ? 'cancelled' : 'failed'
-            current['error'] = error.class.name
+            current['error'] = failure_code(error)
           end
           current['updated_at'] = Time.current.to_f
           write(current)