proper streaming plus tests

Author: amadeovezz

vulcan/lib/remote_manager.rb

@@ -41,18 +41,56 @@ def read_file_to_memory(remote_file_path)
       invoke_ssh_command(command.to_s)[:stdout]
     end
 
-    def stream_file_simple(remote_path, chunk: 64 * 1024)
-      @ssh_pool.with_conn do |ssh|
-        ssh.sftp.file.open(remote_path, 'r') do |file|
-          loop do
-            data = file.read(chunk)
-            break if data.nil? || data.empty?
-            yield data
+    def stream_file_simple(remote_path, chunk: 256 * 1024, max_duration: 300)
+      bytes_read = 0
+      start_time = Time.now
+      
+      # Wrap the entire SFTP operation in a timeout to prevent hanging
+      Timeout.timeout(max_duration) do
+        @ssh_pool.with_conn do |ssh|
+          sftp = ssh.sftp
+          sftp.file.open(remote_path, 'r') do |file|
+            loop do
+              # Check timeout
+              elapsed = Time.now - start_time
+              if elapsed > max_duration
+                Vulcan.instance.logger.warn(
+                  "Stream timeout after #{elapsed.round(1)}s, #{bytes_read} bytes read from #{remote_path}"
+                )
+                raise Timeout::Error, "Stream exceeded #{max_duration}s limit"
+              end
+              
+              data = file.read(chunk)
+              break if data.nil? || data.empty?
+              
+              bytes_read += data.bytesize
+              yield data
+              
+              # Log progress every 100MB for large files
+              if bytes_read % (100 * 1024 * 1024) < chunk
+                Vulcan.instance.logger.info(
+                  "Streaming progress: #{(bytes_read / 1024.0 / 1024.0).round(1)}MB from #{remote_path}"
+                )
+              end
+            end
           end
         end
       end
+    rescue Timeout::Error => e
+      Vulcan.instance.logger.error("Stream timeout after #{(Time.now - start_time).round(1)}s, #{bytes_read} bytes read")
+      raise "Stream timeout: #{e.message}"
     rescue Net::SFTP::StatusException => e
+      Vulcan.instance.logger.error("SFTP error: #{e.description}")
       raise "Remote file error: #{e.description}"
+    rescue IOError, Errno::EPIPE => e
+      # Client likely disconnected
+      Vulcan.instance.logger.warn(
+        "Client disconnected after #{bytes_read} bytes: #{e.message}"
+      )
+      # Don't re-raise - stream was interrupted by client
+    rescue Net::SSH::Disconnect, Errno::ECONNRESET => e
+      Vulcan.instance.logger.error("SSH connection error: #{e.message}")
+      raise "SSH connection lost during streaming: #{e.message}"
     end
 
     def touch(remote_file_path)

vulcan/lib/server/controllers/vulcan_v2_controller.rb

@@ -442,14 +442,20 @@ def stream_download_file
     file_path = File.join(Vulcan::Path.workspace_output_dir(workspace.path), file_name)
     raise Etna::BadRequest, "File not found" unless @remote_manager.file_exists?(file_path)
 
-    response['Content-Type'] = 'application/octet-stream'
-    response['Content-Disposition'] = "attachment; filename=#{File.basename(file_name)}"
-    response['Cache-Control'] = 'no-cache'
-    response.delete_header('Content-Length') if response.respond_to?(:delete_header)
-
-    response.body = Enumerator.new do |yielder|
-      @remote_manager.stream_file_simple(file_path) { |chunk| yielder << chunk }
+    # Create a lazy streaming body that yields chunks on demand
+    # This streams to the frontend without loading the entire file into memory
+    streaming_body = Enumerator.new do |yielder|
+      @remote_manager.stream_file_simple(file_path) do |chunk|
+        yielder << chunk
+      end
     end
+    
+    # Return response with streaming body
+    [200, {
+      'Content-Type' => 'application/octet-stream',
+      'Content-Disposition' => "attachment; filename=#{File.basename(file_name)}",
+      'Cache-Control' => 'no-cache'
+    }, streaming_body]
   end
 
   def cluster_latency

vulcan/spec/spec_helper.rb

@@ -151,6 +151,23 @@ def cp_file(src, dest)
     invoke_ssh_command(command)
   end
 
+  def create_large_file(remote_file_path, size_mb:)
+    Vulcan.instance.logger.info("Creating #{size_mb}MB file at #{remote_file_path}...")
+    
+    # Create parent directory first
+    command = build_command.add('mkdir', '-p', File.dirname(remote_file_path))
+    invoke_ssh_command(command.to_s)
+    
+    Vulcan.instance.logger.info("Generating file content with dd...")
+    # Use dd to create a file with random content of specified size
+    # bs=1M count=size_mb creates a file of size_mb megabytes
+    dd_command = "dd if=/dev/urandom of=#{Shellwords.escape(remote_file_path)} bs=1M count=#{size_mb} 2>/dev/null"
+    invoke_ssh_command(dd_command, timeout: 30)
+    
+    Vulcan.instance.logger.info("Large file created successfully")
+    remote_file_path
+  end
+
 end
 
 

vulcan/spec/workflow_v2_spec.rb

@@ -1360,17 +1360,52 @@ def run_entire_test_workflow(workspace)
 
       get "/api/v2/#{PROJECT}/workspace/#{workspace.id}/file/stream-download/poem.txt"
 
-      # --- force the body to be consumed (two equivalent ways) ---
+      # Force the body to be consumed
       streamed = +""
-      last_response.body.each { |chunk| streamed << chunk }      # explicit
-      # streamed = last_response.body                            # implicit join
+      last_response.body.each { |chunk| streamed << chunk }
 
       expect(last_response.status).to eq(200)
 
-      # headers: rack sets canonical case
+      # Verify headers
       expect(last_response.headers['Content-Type']).to eq('application/octet-stream')
       expect(last_response.headers['Content-Disposition']).to eq('attachment; filename=poem.txt')
       expect(last_response.headers['Cache-Control']).to eq('no-cache')
+      
+      # Verify content matches the expected file
+      expect(streamed).to_not be_empty
+      expect(streamed).to eq(poem_1_text)
+    end
+
+    it 'streams larger files in multiple chunks' do
+      auth_header(:editor)
+
+      workspace = Vulcan::Workspace.first
+      file_name = "large_file.txt"
+      file_path = "#{workspace.path}/output/#{file_name}"
+      
+      # Create a 1MB file directly on the remote server (bypassing echo limits)
+      remote_manager.create_large_file(file_path, size_mb: 1)
+      
+      # Get the actual file size for verification
+      stat_result = remote_manager.invoke_ssh_command("stat -c %s #{file_path}")
+      expected_size = stat_result[:stdout].strip.to_i
+
+      get "/api/v2/#{PROJECT}/workspace/#{workspace.id}/file/stream-download/#{file_name}"
+
+      expect(last_response.status).to eq(200)
+      
+      # Rack::Test joins all body chunks into a single string
+      # The server streams the file in 256KB chunks (4 chunks for 1MB file)
+      streamed = last_response.body
+      
+      # Verify the complete size matches (1MB = 1048576 bytes)
+      expect(streamed.bytesize).to eq(expected_size)
+      expect(streamed.bytesize).to eq(1024 * 1024) # 1MB
+      
+      # Verify headers
+      expect(last_response.headers['Content-Type']).to eq('application/octet-stream')
+      expect(last_response.headers['Content-Disposition']).to eq("attachment; filename=#{file_name}")
+      expect(last_response.headers['Cache-Control']).to eq('no-cache')
     end
 
     it 'raises an error if the file does not exist' do