Class: StellarCoreCommander::Process

Inherits:
Object
  • Object
show all
Includes:
Contracts
Defined in:
lib/stellar_core_commander/process.rb

Direct Known Subclasses

DockerProcess, LocalProcess

Defined Under Namespace

Classes: AlreadyRunning, Crash

Constant Summary collapse

DEFAULT_HOST = 
'127.0.0.1'
SPECIAL_PEERS = 
{
  :testnet1 => {
    :dns => "core-testnet1.stellar.org",
    :key => "GDKXE2OZMJIPOSLNA6N6F2BVCI3O777I2OOC4BV7VOYUEHYX7RTRYA7Y",
    :name => "core-testnet-001",
    :get => "wget -q https://s3-eu-west-1.amazonaws.com/history.stellar.org/prd/core-testnet/%s/{0} -O {1}"
  },
  :testnet2 => {
    :dns => "core-testnet2.stellar.org",
    :key => "GCUCJTIYXSOXKBSNFGNFWW5MUQ54HKRPGJUTQFJ5RQXZXNOLNXYDHRAP",
    :name => "core-testnet-002",
    :get => "wget -q https://s3-eu-west-1.amazonaws.com/history.stellar.org/prd/core-testnet/%s/{0} -O {1}"
  },
  :testnet3 => {
    :dns => "core-testnet3.stellar.org",
    :key => "GC2V2EFSXN6SQTWVYA5EPJPBWWIMSD2XQNKUOHGEKB535AQE2I6IXV2Z",
    :name => "core-testnet-003",
    :get => "wget -q https://s3-eu-west-1.amazonaws.com/history.stellar.org/prd/core-testnet/%s/{0} -O {1}"
  }
}

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(params) ⇒ Process

Returns a new instance of Process.



80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/stellar_core_commander/process.rb', line 80

def initialize(params)
  #config
  @transactor         = params[:transactor]
  @working_dir        = params[:working_dir]
  @name               = params[:name]
  @base_port          = params[:base_port]
  @identity           = params[:identity]
  @quorum             = params[:quorum]
  @peers              = params[:peers] || params[:quorum]
  @manual_close       = params[:manual_close] || false
  @await_sync         = params.fetch(:await_sync, true)
  @accelerate_time    = params[:accelerate_time] || false
  @catchup_complete   = params[:catchup_complete] || false
  @forcescp           = params.fetch(:forcescp, true)
  @validate           = params.fetch(:validate, true)
  @host               = params[:host]
  @atlas              = params[:atlas]
  @atlas_interval     = params[:atlas_interval]
  @use_s3             = params[:use_s3]
  @s3_history_region  = params[:s3_history_region]
  @s3_history_prefix  = params[:s3_history_prefix]
  @database_url       = params[:database_url]
  @keep_database      = params[:keep_database]
  @debug              = params[:debug]
  @network_passphrase = params[:network_passphrase] || Stellar::Networks::TESTNET

  # state
  @unverified   = []

  if not @quorum.include? @name
    @quorum = @quorum + [@name]
  end

  if not @peers.include? @name
    @peers = @peers + [@name]
  end

  @server = Faraday.new(url: "http://#{hostname}:#{http_port}") do |conn|
    conn.request :url_encoded
    conn.adapter Faraday.default_adapter
  end
end

Instance Attribute Details

#atlas ⇒ Object (readonly)

Returns the value of attribute atlas.



28
29
30
 
# File 'lib/stellar_core_commander/process.rb', line 28

def atlas
  @atlas
end

#atlas_interval ⇒ Object (readonly)

Returns the value of attribute atlas_interval.



29
30
31
 
# File 'lib/stellar_core_commander/process.rb', line 29

def atlas_interval
  @atlas_interval
end

#base_port ⇒ Object (readonly)

Returns the value of attribute base_port.



23
24
25
 
# File 'lib/stellar_core_commander/process.rb', line 23

def base_port
  @base_port
end

#host ⇒ Object (readonly)

Returns the value of attribute host.



27
28
29
 
# File 'lib/stellar_core_commander/process.rb', line 27

def host
  @host
end

#identity ⇒ Object (readonly)

Returns the value of attribute identity.



24
25
26
 
# File 'lib/stellar_core_commander/process.rb', line 24

def identity
  @identity
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



22
23
24
 
# File 'lib/stellar_core_commander/process.rb', line 22

def name
  @name
end

#network_passphrase ⇒ Object (readonly)

Returns the value of attribute network_passphrase.



30
31
32
 
# File 'lib/stellar_core_commander/process.rb', line 30

def network_passphrase
  @network_passphrase
end

#server ⇒ Object (readonly)

Returns the value of attribute server.



25
26
27
 
# File 'lib/stellar_core_commander/process.rb', line 25

def server
  @server
end

#transactor ⇒ Object (readonly)

Returns the value of attribute transactor.



20
21
22
 
# File 'lib/stellar_core_commander/process.rb', line 20

def transactor
  @transactor
end

#unverified ⇒ Object

Returns the value of attribute unverified.



26
27
28
 
# File 'lib/stellar_core_commander/process.rb', line 26

def unverified
  @unverified
end

#working_dir ⇒ Object (readonly)

Returns the value of attribute working_dir.



21
22
23
 
# File 'lib/stellar_core_commander/process.rb', line 21

def working_dir
  @working_dir
end

Instance Method Details

#account_count ⇒ Object 



489
490
491
 
# File 'lib/stellar_core_commander/process.rb', line 489

def 
  database.fetch("SELECT count(*) FROM accounts").first[:count]
end

#account_row(account) ⇒ Object 



447
448
449
450
451
 
# File 'lib/stellar_core_commander/process.rb', line 447

def ()
  row = database[:accounts].where(:accountid => .address).first
  raise "Missing account in #{idname}'s database: #{.address}" unless row
  row
end

#await_sync? ⇒ Boolean

Returns:

  • (Boolean) 


307
308
309
 
# File 'lib/stellar_core_commander/process.rb', line 307

def await_sync?
  @await_sync
end

#balance_for(account) ⇒ Object 



459
460
461
 
# File 'lib/stellar_core_commander/process.rb', line 459

def balance_for()
  ( )[:balance]
end

#catchup(ledger, mode) ⇒ Object 



273
274
275
 
# File 'lib/stellar_core_commander/process.rb', line 273

def catchup(ledger, mode)
  server.get("/catchup?ledger=#{ledger}&mode=#{mode}")
end

#check_equal(kind, x, y) ⇒ Object

Raises:



519
520
521
 
# File 'lib/stellar_core_commander/process.rb', line 519

def check_equal(kind, x, y)
  raise UnexpectedDifference.new(kind, x, y) if x != y
end

#check_equal_ledger_objects(other) ⇒ Object 



524
525
526
527
528
529
530
531
532
 
# File 'lib/stellar_core_commander/process.rb', line 524

def check_equal_ledger_objects(other)
  check_equal "account count", , other.
  check_equal "trustline count", trustline_count, other.trustline_count
  check_equal "offer count", offer_count, other.offer_count

  check_equal "ten accounts", ten_accounts, other.ten_accounts
  check_equal "ten trustlines", ten_trustlines, other.ten_trustlines
  check_equal "ten offers", ten_offers, other.ten_offers
end

#check_ledger_sequence_is_prefix_of(other) ⇒ Object 



535
536
537
538
539
540
541
542
543
544
545
546
547
 
# File 'lib/stellar_core_commander/process.rb', line 535

def check_ledger_sequence_is_prefix_of(other)
  q = "SELECT ledgerseq, ledgerhash FROM ledgerheaders ORDER BY ledgerseq"
  our_headers = other.database.fetch(q).all
  other_headers = other.database.fetch(q).all
  our_hash = {}
  other_hash = {}
  other_headers.each do |row|
    other_hash[row[:ledgerseq]] = row[:ledgerhash]
  end
  our_headers.each do |row|
    check_equal "ledger hashes", other_hash[row[:ledgerseq]], row[:ledgerhash]
  end
end

#check_no_error_metrics ⇒ Object 



372
373
374
375
376
377
378
379
380
381
382
383
 
# File 'lib/stellar_core_commander/process.rb', line 372

def check_no_error_metrics
  m = metrics
  for metric in ["scp.envelope.invalidsig",
                 "history.publish.failure",
                 "history.catchup.failure"]
    c = m[metric]["count"] rescue 0
    if c != 0
      raise "nonzero metrics count for #{metric}: #{c}"
    end
  end
  true
end

#checkdb_runs ⇒ Object 



416
417
418
 
# File 'lib/stellar_core_commander/process.rb', line 416

def checkdb_runs
  metrics_count "bucket.checkdb.execute"
end

#close_ledger ⇒ Object 



245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
 
# File 'lib/stellar_core_commander/process.rb', line 245

def close_ledger
  prev_ledger = latest_ledger
  next_ledger = prev_ledger + 1

  Timeout.timeout(close_timeout) do

    server.get("manualclose") if manual_close?

    loop do
      current_ledger = latest_ledger

      case
      when current_ledger == next_ledger
        break
      when current_ledger > next_ledger
        raise "#{idname} jumped two ledgers, from #{prev_ledger} to #{current_ledger}"
      else
        $stderr.puts "#{idname} waiting for ledger #{next_ledger} (current: #{current_ledger}, ballots prepared: #{scp_ballots_prepared})"
        sleep 0.5
      end
    end
  end
  $stderr.puts "#{idname} closed #{latest_ledger}"

  true
end

#close_timeout ⇒ Object 



562
563
564
 
# File 'lib/stellar_core_commander/process.rb', line 562

def close_timeout
  150.0
end

#crashed? ⇒ Boolean

Returns:

  • (Boolean) 


607
608
609
 
# File 'lib/stellar_core_commander/process.rb', line 607

def crashed?
  launched? && stopped?
end

#database ⇒ Object 



199
200
201
 
# File 'lib/stellar_core_commander/process.rb', line 199

def database
  @database ||= Sequel.connect(database_url)
end

#database_host ⇒ Object 



189
190
191
 
# File 'lib/stellar_core_commander/process.rb', line 189

def database_host
  database_uri.host
end

#database_name ⇒ Object 



194
195
196
 
# File 'lib/stellar_core_commander/process.rb', line 194

def database_name
  database_uri.path[1..-1]
end

#database_password ⇒ Object 



209
210
211
 
# File 'lib/stellar_core_commander/process.rb', line 209

def database_password
  database_uri.password
end

#database_port ⇒ Object 



214
215
216
 
# File 'lib/stellar_core_commander/process.rb', line 214

def database_port
  database_uri.port || "5432"
end

#database_uri ⇒ Object 



184
185
186
 
# File 'lib/stellar_core_commander/process.rb', line 184

def database_uri
  URI.parse(database_url)
end

#database_url ⇒ Object 



175
176
177
178
179
180
181
 
# File 'lib/stellar_core_commander/process.rb', line 175

def database_url
  if @database_url.present?
    @database_url.strip
  else
    default_database_url
  end
end

#database_user ⇒ Object 



204
205
206
 
# File 'lib/stellar_core_commander/process.rb', line 204

def database_user
  database_uri.user
end

#db_store_state(name) ⇒ Object 



469
470
471
 
# File 'lib/stellar_core_commander/process.rb', line 469

def db_store_state(name)
  database.select(:state).from(:storestate).filter(statename: name).first[:state]
end

#dsn ⇒ Object 



219
220
221
222
223
224
225
226
 
# File 'lib/stellar_core_commander/process.rb', line 219

def dsn
  base = "postgresql://dbname=#{database_name} "
  base << " user=#{database_user}" if database_user.present?
  base << " password=#{database_password}" if database_password.present?
  base << " host=#{database_host} port=#{database_port}" if database_host.present?

  base
end

#dump_database ⇒ Object

Raises:

  • (NotImplementedError) 


629
630
631
 
# File 'lib/stellar_core_commander/process.rb', line 629

def dump_database
  raise NotImplementedError, "implement in subclass"
end

#dump_info ⇒ Object 



352
353
354
 
# File 'lib/stellar_core_commander/process.rb', line 352

def dump_info
  dump_server_query("info")
end

#dump_metrics ⇒ Object 



347
348
349
 
# File 'lib/stellar_core_commander/process.rb', line 347

def dump_metrics
  dump_server_query("metrics")
end

#dump_scp_state ⇒ Object 



357
358
359
 
# File 'lib/stellar_core_commander/process.rb', line 357

def dump_scp_state
  dump_server_query("scp")
end

#dump_server_query(s) ⇒ Object 



337
338
339
340
341
342
343
344
 
# File 'lib/stellar_core_commander/process.rb', line 337

def dump_server_query(s)
  fname = "#{working_dir}/#{s}-#{Time.now.to_i}-#{rand 100000}.json"
  $stderr.puts "dumping server query #{fname}"
  response = server.get("/#{s}")
  File.open(fname, 'w') {|f| f.write(response.body) }
rescue
  nil
end

#has_special_peers? ⇒ Boolean

Returns:

  • (Boolean) 


124
125
126
 
# File 'lib/stellar_core_commander/process.rb', line 124

def has_special_peers?
  @peers.any? {|q| SPECIAL_PEERS.has_key? q}
end

#history_archive_state ⇒ Object 



484
485
486
 
# File 'lib/stellar_core_commander/process.rb', line 484

def history_archive_state
  ActiveSupport::JSON.decode(db_store_state("historyarchivestate"))
end

#hostname ⇒ Object 



557
558
559
 
# File 'lib/stellar_core_commander/process.rb', line 557

def hostname
  host || DEFAULT_HOST
end

#http_port ⇒ Object 



426
427
428
 
# File 'lib/stellar_core_commander/process.rb', line 426

def http_port
  base_port
end

#idname ⇒ Object 



170
171
172
 
# File 'lib/stellar_core_commander/process.rb', line 170

def idname
  "#{@name}-#{@base_port}-#{@identity.address[0..5]}"
end

#info ⇒ Object 



278
279
280
281
282
283
284
 
# File 'lib/stellar_core_commander/process.rb', line 278

def info
  response = server.get("/info")
  body = ActiveSupport::JSON.decode(response.body)
  body["info"]
rescue
  {}
end

#info_field(k) ⇒ Object 



287
288
289
290
291
292
 
# File 'lib/stellar_core_commander/process.rb', line 287

def info_field(k)
  i = info
  i[k]
rescue
  false
end

#latest_ledger ⇒ Object 



464
465
466
 
# File 'lib/stellar_core_commander/process.rb', line 464

def latest_ledger
  database[:ledgerheaders].max(:ledgerseq)
end

#latest_ledger_hash ⇒ Object 



474
475
476
477
478
479
480
481
 
# File 'lib/stellar_core_commander/process.rb', line 474

def latest_ledger_hash
  s_lcl = db_store_state("lastclosedledger")
  t_lcl = database.select(:ledgerhash)
    .from(:ledgerheaders)
    .filter(:ledgerseq=>latest_ledger).first[:ledgerhash]
  raise "inconsistent last-ledger hashes in db: #{t_lcl} vs. #{s_lcl}" if t_lcl != s_lcl
  s_lcl
end

#launched? ⇒ Boolean

Returns:

  • (Boolean) 


602
603
604
 
# File 'lib/stellar_core_commander/process.rb', line 602

def launched?
  !!@launched
end

#ledger_num ⇒ Object 



300
301
302
303
304
 
# File 'lib/stellar_core_commander/process.rb', line 300

def ledger_num
  (info_field "ledger")["num"]
rescue
  0
end

#load_generation_runs ⇒ Object 



391
392
393
 
# File 'lib/stellar_core_commander/process.rb', line 391

def load_generation_runs
  metrics_count "loadgen.run.complete"
end

#manual_close? ⇒ Boolean

Returns:

  • (Boolean) 


240
241
242
 
# File 'lib/stellar_core_commander/process.rb', line 240

def manual_close?
  @manual_close
end

#metrics ⇒ Object 



312
313
314
315
316
317
318
 
# File 'lib/stellar_core_commander/process.rb', line 312

def metrics
  response = server.get("/metrics")
  body = ActiveSupport::JSON.decode(response.body)
  body["metrics"]
rescue
  {}
end

#metrics_1m_rate(k) ⇒ Object 



329
330
331
332
333
334
 
# File 'lib/stellar_core_commander/process.rb', line 329

def metrics_1m_rate(k)
  m = metrics
  m[k]["1_min_rate"]
rescue
  0
end

#metrics_count(k) ⇒ Object 



321
322
323
324
325
326
 
# File 'lib/stellar_core_commander/process.rb', line 321

def metrics_count(k)
  m = metrics
  m[k]["count"]
rescue
  0
end

#node_map_or_special_field(nodes, field, include_self) ⇒ Object 



129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/stellar_core_commander/process.rb', line 129

def node_map_or_special_field(nodes, field, include_self)
  specials = nodes.select {|q| SPECIAL_PEERS.has_key? q}
  if specials.empty?
    (nodes.map do |q|
      if q != @name or include_self
        yield q
      end
    end).compact
  else
    specials.map {|q| SPECIAL_PEERS[q][field]}
  end
end

#objects_checked ⇒ Object 



421
422
423
 
# File 'lib/stellar_core_commander/process.rb', line 421

def objects_checked
  metrics_count "bucket.checkdb.object-compare"
end

#offer_count ⇒ Object 



499
500
501
 
# File 'lib/stellar_core_commander/process.rb', line 499

def offer_count
  database.fetch("SELECT count(*) FROM offers").first[:count]
end

#operations_per_second ⇒ Object 



406
407
408
 
# File 'lib/stellar_core_commander/process.rb', line 406

def operations_per_second
  metrics_1m_rate "transaction.op.apply"
end

#peer_connections ⇒ Object 



157
158
159
160
161
162
 
# File 'lib/stellar_core_commander/process.rb', line 157

def peer_connections
  node_map_or_special_field @peers, :dns, false do |q|
    p = @transactor.get_process(q)
    "#{p.hostname}:#{p.peer_port}"
  end
end

#peer_names ⇒ Object 



150
151
152
153
154
 
# File 'lib/stellar_core_commander/process.rb', line 150

def peer_names
  node_map_or_special_field @peers, :name, true do |q|
    q.to_s
  end
end

#peer_port ⇒ Object 



431
432
433
 
# File 'lib/stellar_core_commander/process.rb', line 431

def peer_port
  base_port + 1
end

#prepare ⇒ Object 



612
613
614
615
 
# File 'lib/stellar_core_commander/process.rb', line 612

def prepare
  # noop by default, implement in subclass to customize behavior
  nil
end

#quorum ⇒ Object 



143
144
145
146
147
 
# File 'lib/stellar_core_commander/process.rb', line 143

def quorum
  node_map_or_special_field @quorum, :key, @validate do |q|
    @transactor.get_process(q).identity.address
  end
end

#required_ports ⇒ Object 



165
166
167
 
# File 'lib/stellar_core_commander/process.rb', line 165

def required_ports
  2
end

#run ⇒ Object

Raises:



618
619
620
621
622
623
624
625
 
# File 'lib/stellar_core_commander/process.rb', line 618

def run
  raise Process::AlreadyRunning, "already running!" if running?
  raise Process::Crash, "process #{name} has crashed. cannot run process again" if crashed?

  setup
  launch_process
  @launched = true
end

#run_cmd(cmd, args) ⇒ Object 



585
586
587
588
589
590
591
592
593
594
 
# File 'lib/stellar_core_commander/process.rb', line 585

def run_cmd(cmd, args)
  args += [{
      out: ["stellar-core.log", "a"],
      err: ["stellar-core.log", "a"],
    }]

  Dir.chdir working_dir do
    system(cmd, *args)
  end
end

#scp_ballots_prepared ⇒ Object 



362
363
364
 
# File 'lib/stellar_core_commander/process.rb', line 362

def scp_ballots_prepared
  metrics_count "scp.ballot.prepare"
end

#scp_quorum_heard ⇒ Object 



367
368
369
 
# File 'lib/stellar_core_commander/process.rb', line 367

def scp_quorum_heard
  metrics_count "scp.quorum.heard"
end

#sequence_for(account) ⇒ Object 



454
455
456
 
# File 'lib/stellar_core_commander/process.rb', line 454

def sequence_for()
  ( )[:seqnum]
end

#start_checkdb ⇒ Object 



411
412
413
 
# File 'lib/stellar_core_commander/process.rb', line 411

def start_checkdb
  server.get("/checkdb")
end

#start_load_generation(accounts, txs, txrate) ⇒ Object 



386
387
388
 
# File 'lib/stellar_core_commander/process.rb', line 386

def start_load_generation(accounts, txs, txrate)
  server.get("/generateload?accounts=#{accounts}&txs=#{txs}&txrate=#{txrate}")
end

#stopped? ⇒ Boolean

Returns:

  • (Boolean) 


597
598
599
 
# File 'lib/stellar_core_commander/process.rb', line 597

def stopped?
  !running?
end

#submit_transaction(envelope_hex) ⇒ Object 



436
437
438
439
440
441
442
443
444
 
# File 'lib/stellar_core_commander/process.rb', line 436

def submit_transaction(envelope_hex)
  response = server.get("tx", blob: envelope_hex)
  body = ActiveSupport::JSON.decode(response.body)

  if body["status"] == "ERROR"
    raise "transaction on #{idname} failed: #{body.inspect}"
  end

end

#sync_timeout ⇒ Object 



567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
 
# File 'lib/stellar_core_commander/process.rb', line 567

def sync_timeout
  if has_special_peers? and @catchup_complete
    # catchup-complete can take quite a while on testnet; for now,
    # give such tests an hour. May require a change in strategy later.
    3600.0
  else
    # Checkpoints are made every 64 ledgers = 320s on a normal network,
    # or every 8 ledgers = 8s on an accelerated-time network; we give you
    # 3 checkpoints to make it to a sync (~16min) before giving up. The
    # accelerated-time variant tends to need more tries due to S3 not
    # admitting writes instantaneously, so we do not use a tighter bound
    # for that case, just use the same 16min value, despite commonly
    # succeeding in 20s or less.
    320.0 * 3
  end
end

#synced? ⇒ Boolean

Returns:

  • (Boolean) 


295
296
297
 
# File 'lib/stellar_core_commander/process.rb', line 295

def synced?
  (info_field "state") == "Synced!"
end

#ten_accounts ⇒ Object 



504
505
506
 
# File 'lib/stellar_core_commander/process.rb', line 504

def ten_accounts
  database.fetch("SELECT * FROM accounts ORDER BY accountid LIMIT 10").all
end

#ten_offers ⇒ Object 



509
510
511
 
# File 'lib/stellar_core_commander/process.rb', line 509

def ten_offers
  database.fetch("SELECT * FROM offers ORDER BY sellerid LIMIT 10").all
end

#ten_trustlines ⇒ Object 



514
515
516
 
# File 'lib/stellar_core_commander/process.rb', line 514

def ten_trustlines
  database.fetch("SELECT * FROM trustlines ORDER BY accountid, issuer, assetcode LIMIT 10").all
end

#transaction_result(hex_hash) ⇒ Object 



550
551
552
553
554
 
# File 'lib/stellar_core_commander/process.rb', line 550

def transaction_result(hex_hash)
  row = database[:txhistory].where(txid:hex_hash).first
  return if row.blank?
  row[:txresult]
end

#transactions_applied ⇒ Object 



396
397
398
 
# File 'lib/stellar_core_commander/process.rb', line 396

def transactions_applied
  metrics_count "ledger.transaction.apply"
end

#transactions_per_second ⇒ Object 



401
402
403
 
# File 'lib/stellar_core_commander/process.rb', line 401

def transactions_per_second
  metrics_1m_rate "ledger.transaction.apply"
end

#trustline_count ⇒ Object 



494
495
496
 
# File 'lib/stellar_core_commander/process.rb', line 494

def trustline_count
  database.fetch("SELECT count(*) FROM trustlines").first[:count]
end

#wait_for_ready ⇒ Object 



229
230
231
232
233
234
235
236
237
 
# File 'lib/stellar_core_commander/process.rb', line 229

def wait_for_ready
  Timeout.timeout(sync_timeout) do
    loop do
      break if synced?
      $stderr.puts "waiting until stellar-core #{idname} is synced (state: #{info_field 'state'}, quorum heard: #{scp_quorum_heard})"
      sleep 1
    end
  end
end