Module: PgPipeline::TransactionOps

Defined in:
lib/pg_pipeline/transaction.rb

Class Method Summary collapse

Class Method Details

.commit(conn) ⇒ Object 



51
52
53
54
55
56
57
58
59
60
 
# File 'lib/pg_pipeline/transaction.rb', line 51

def commit(conn)
  conn.exec("COMMIT")
rescue PG::Error => error
  raise unless indeterminate_commit_failure?(conn, error)

  raise IndeterminateCommitError.new(
    "COMMIT acknowledgement was not received; the transaction may have committed " \
    "and must not be retried blindly (#{error.class}: #{error.message})"
  ), cause: error
end

.indeterminate_commit_failure?(conn, error) ⇒ Boolean

Returns:

  • (Boolean) 


62
63
64
65
66
67
68
69
70
71
 
# File 'lib/pg_pipeline/transaction.rb', line 62

def indeterminate_commit_failure?(conn, error)
  return true if defined?(PG::ConnectionBad) && error.is_a?(PG::ConnectionBad)
  return true if conn.finished?

  conn.status != PG::CONNECTION_OK
rescue PG::Error
  true
rescue NoMethodError
  false
end

.rollback_quietly(tx) ⇒ Object 



103
104
105
106
107
108
109
110
111
112
 
# File 'lib/pg_pipeline/transaction.rb', line 103

def rollback_quietly(tx)
  return unless tx.open?

  conn = SessionOps.connection(tx)
  conn&.exec("ROLLBACK")
rescue PG::Error
  nil
ensure
  tx.__send__(:open=, false)
end

.rollback_to_savepoint(tx, ident) ⇒ Object 



93
94
95
96
97
98
99
100
101
 
# File 'lib/pg_pipeline/transaction.rb', line 93

def rollback_to_savepoint(tx, ident)
  conn = SessionOps.connection(tx)
  return unless conn

  conn.exec("ROLLBACK TO SAVEPOINT #{ident}")
  conn.exec("RELEASE SAVEPOINT #{ident}")
rescue PG::Error
  nil
end

.run(tx) ⇒ Object

Raises:



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/pg_pipeline/transaction.rb', line 33

def run(tx)
  SessionOps.ensure_active!(tx)
  raise Error, "transaction is already open" if tx.open?

  conn = SessionOps.connection(tx)
  conn.exec("BEGIN")
  tx.__send__(:open=, true)
  begin
    result = yield tx
    commit(conn)
    tx.__send__(:open=, false)
    result
  rescue Exception
    rollback_quietly(tx)
    raise
  end
end

.savepoint(tx, name) ⇒ Object

Raises:



73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/pg_pipeline/transaction.rb', line 73

def savepoint(tx, name)
  SessionOps.ensure_active!(tx)
  raise Error, "savepoint requires an open transaction" unless tx.open?

  conn = SessionOps.connection(tx)
  seq = tx.__send__(:savepoint_seq) + 1
  tx.__send__(:savepoint_seq=, seq)
  point = name || "pgp_sp_#{seq}"
  ident = conn.quote_ident(point)
  conn.exec("SAVEPOINT #{ident}")
  begin
    result = yield tx
    conn.exec("RELEASE SAVEPOINT #{ident}")
    result
  rescue Exception
    rollback_to_savepoint(tx, ident)
    raise
  end
end