Class: NoopBackup::Commands::Backup

Inherits:
Object
  • Object
show all
Defined in:
lib/noop_backup/commands/backup.rb

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Backup

Returns a new instance of Backup.



37
38
39
40
41
 
# File 'lib/noop_backup/commands/backup.rb', line 37

def initialize
  @key = generate_key
  @store_results = []
  @sinks = []
end

Class Method Details

.execute(report: NoopBackup.config.report?) ⇒ Object 



29
30
31
32
33
34
35
 
# File 'lib/noop_backup/commands/backup.rb', line 29

def self.execute(report: NoopBackup.config.report?)
  result = new.execute

  result.report if report

  result
end

Instance Method Details

#execute ⇒ Object 



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/noop_backup/commands/backup.rb', line 43

def execute
  perform_sanity_check!

  commands = [config.dump_command]

  # Pipe pg_dump through pv if installed for a basic progress report
  commands << ["pv", "-btra"] if config.report? && system("which", "pv", out: File::NULL, err: File::NULL)

  Open3.pipeline_r(*commands) do |last_stdout, wait_threads|
    @sinks = config.stores.map do |store|
      reader, writer = IO.pipe(binmode: true)

      thread = Thread.new do
        store.backup!(@key, reader)
      rescue => e
        # *always* return a Result, even if unexpected. Add store name for debugging.
        NoopBackup::Stores::Result.new(success: false, error: e, store: store.class.name, key: @key)
      ensure
        reader.close
      end

      NoopBackup::Tee::Sink.new(store:, writer:, thread:)
    end

    sinks_fanout = NoopBackup::Tee.new(@sinks)

    begin
      IO.copy_stream(last_stdout, sinks_fanout)
    rescue => e
      raise NoopBackup::DumpFailedError, "streaming failed: #{e.message}"
    ensure
      @sinks.each(&:close)
    end

    @store_results = @sinks.map(&:collect)

    raise NoopBackup::DumpFailedError, "pipeline failed" unless wait_threads.all? { |t| t.value.success? }
  end

  CommandResult.new(store_results: @store_results)
rescue NoopBackup::DumpFailedError => error
  # The dump stream itself failed, so uploads that finished are truncated copies of a bad stream — delete them.
  # Collect the results first: if copy_stream raised, the collection line above never ran. Sinks are already
  # closed on every DumpFailedError path, so collect can't block.
  @store_results = @sinks.map(&:collect)

  cleanup_uploaded_stores!

  CommandResult.new(error:, store_results: @store_results)
rescue => error
  CommandResult.new(error:, store_results: @store_results)
end