lib/zold/node/async_entrance.rb in zold-0.24.3 vs lib/zold/node/async_entrance.rb in zold-0.24.4

@@ -62,11 +62,15 @@
       FileUtils.mkdir_p(@dir)
       DirItems.new(@dir).fetch.each do |f|
         file = File.join(@dir, f)
         if /^[0-9a-f]{16}-/.match?(f)
           id = f.split('-')[0]
-          @queue << { id: Id.new(id), file: file }
+          if exists?(id, IO.read(file))
+            @queue << { id: Id.new(id), file: file }
+          else
+            File.delete(file)
+          end
         else
           File.delete(file)
         end
       end
       @log.info("#{@queue.size} wallets pre-loaded into async_entrace from #{@dir}") unless @queue.size.zero?
@@ -93,23 +97,40 @@
           SoftError,
           "Queue is too long (#{@queue.size} wallets), can't add #{id}/#{Size.new(body.length)}, try again later"
         )
       end
       start = Time.now
-      loop do
-        uuid = SecureRandom.uuid
-        file = File.join(@dir, "#{id}-#{uuid}")
-        next if File.exist?(file)
-        IO.write(file, body)
-        @queue << { id: id, file: file }
-        @log.debug("Added #{id}/#{Size.new(body.length)} to the queue at pos.#{@queue.size} \
-in #{Age.new(start, limit: 0.05)}")
-        break
+      unless exists?(id, body)
+        loop do
+          uuid = SecureRandom.uuid
+          file = File.join(@dir, "#{id}-#{uuid}")
+          next if File.exist?(file)
+          IO.write(file, body)
+          @queue << { id: id, file: file }
+          @log.debug("Added #{id}/#{Size.new(body.length)} to the queue at pos.#{@queue.size} \
+  in #{Age.new(start, limit: 0.05)}")
+          break
+        end
       end
       [id]
     end
 
     private
+
+    # Returns TRUE if a file for this wallet is already in the queue.
+    def exists?(id, body)
+      DirItems.new(@dir).fetch.each do |f|
+        next unless f.start_with?("#{id}-")
+        return true if safe_read(File.join(@dir, f)) == body
+      end
+      false
+    end
+
+    def safe_read(file)
+      IO.read(file)
+    rescue Errno::ENOENT
+      ''
+    end
 
     def take
       start = Time.now
       item = @queue.pop
       Thread.current.thread_variable_set(:wallet, item[:id].to_s)