Class: PG::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/tep/pg.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(opts) ⇒ Connection

Returns a new instance of Connection.



188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
 
# File 'lib/tep/pg.rb', line 188

def initialize(opts)
  @pgh = -1
  @last_sqlstate = ""
  @last_error_message = ""
  @last_result_rh = -1
  # Never-connected sentinel: PG::Connection.new("") is the
  # type-seeding shape (see the _tep_seed_pg_* block). Exec on a
  # seed conn returns the empty-Result sentinel instead of
  # raising -- a REAL failed connect (non-empty conninfo) still
  # raises PG::UnableToSend on first exec, per the documented
  # non-raising-connect / raising-exec contract.
  @seed_conn = false
  if opts.is_a?(String) && opts.length == 0
    @seed_conn = true
  end
  if opts.is_a?(String)
    if Tep::Scheduler.scheduled_context?
      h = Connection.async_connect(opts)
    else
      h = Pg.tep_pg_connect(opts)
    end
  else
    # Hash-conninfo form: pack the key/value pairs into NUL-delimited
    # buffers for the C shim. (`opts` narrows to Hash in this
    # is_a?(String) ELSE branch -- the narrowing gap that blocked the
    # re-pin, matz/spinel#1434, is fixed as of the SPINEL_PIN bump.)
    keys = ""
    vals = ""
    n = 0
    opts.each do |k, v|
      keys = keys + k + "\0"
      vals = vals + v + "\0"
      n += 1
    end
    h = Pg.tep_pg_connect_kv(keys, vals, n)
  end
  if h < 0
    # Slot 0 holds the most recent connect-failure error message
    # (PQstatus on a failed PQconnectdb still gives a readable
    # error, but the conn itself is closed by the time we get
    # here -- the shim stashes the message before PQfinish).
    @last_error_message = Pg.tep_pg_error_message(0)
    @last_sqlstate = ""
    # Connection-failure surfaces via `c.last_error_message` +
    # `c.connected?` after the constructor returns -- the
    # constructor stays non-raising on purpose (PG::Pool seeds its
    # free list with `PG::Connection.new("")` before a server is
    # reachable; a raising constructor would blow up at module
    # load). Callers must check `c.connected?` before exec. NB:
    # this is the lone non-raising path -- query methods raise
    # PG::Error subclasses now that spinel supports namespaced
    # raise + rescue (matz/spinel#627 + #1041).
  end
  @pgh = h
end

Instance Attribute Details

#last_error_message ⇒ Object

Error context for the most recent exception raised by this connection. Spinel's raise X.new(msg, ...) lowering doesn't handle custom initializers (#622), so the SQLSTATE / message / owning-result-handle live here instead. Read after rescue PG::Error => e:

begin; conn.exec_params(sql, params)
rescue PG::Error => e
  sqlstate = conn.last_sqlstate
  full_msg = conn.last_error_message
end

AR's translate_exception_class(message, sql, binds) uses e.is_a?(PG::UniqueViolation) etc., which still works -- the class hierarchy is intact; only the per-exception accessors move to the connection.



186
187
188
 
# File 'lib/tep/pg.rb', line 186

def last_error_message
  @last_error_message
end

#last_result_rh ⇒ Object

Error context for the most recent exception raised by this connection. Spinel's raise X.new(msg, ...) lowering doesn't handle custom initializers (#622), so the SQLSTATE / message / owning-result-handle live here instead. Read after rescue PG::Error => e:

begin; conn.exec_params(sql, params)
rescue PG::Error => e
  sqlstate = conn.last_sqlstate
  full_msg = conn.last_error_message
end

AR's translate_exception_class(message, sql, binds) uses e.is_a?(PG::UniqueViolation) etc., which still works -- the class hierarchy is intact; only the per-exception accessors move to the connection.



186
187
188
 
# File 'lib/tep/pg.rb', line 186

def last_result_rh
  @last_result_rh
end

#last_sqlstate ⇒ Object

Error context for the most recent exception raised by this connection. Spinel's raise X.new(msg, ...) lowering doesn't handle custom initializers (#622), so the SQLSTATE / message / owning-result-handle live here instead. Read after rescue PG::Error => e:

begin; conn.exec_params(sql, params)
rescue PG::Error => e
  sqlstate = conn.last_sqlstate
  full_msg = conn.last_error_message
end

AR's translate_exception_class(message, sql, binds) uses e.is_a?(PG::UniqueViolation) etc., which still works -- the class hierarchy is intact; only the per-exception accessors move to the connection.



186
187
188
 
# File 'lib/tep/pg.rb', line 186

def last_sqlstate
  @last_sqlstate
end

#pgh ⇒ Object

:pgh rather than :handle -- same poly-dispatch widening concern as Tep::SQLite#dbh (sharing a method name with Tep::Handler#handle confuses spinel's same-named-imeth-across- classes unifier).



169
170
171
 
# File 'lib/tep/pg.rb', line 169

def pgh
  @pgh
end

Class Method Details

.async_connect(conninfo) ⇒ Object

Drive PQconnectStart + PQconnectPoll, parking on io_wait between poll calls. Returns the conn slot (>=1) on success or -1 on failure. The C shim's tep_pg_connect_poll stashes the libpq error message on a FAILED return so Pg.tep_pg_error_message(0) still surfaces the diagnostic for the Connection.new "connect failed" branch.

libpq's PostgresPollingStatusType:

0 = PGRES_POLLING_FAILED
1 = PGRES_POLLING_READING    (wait for fd READ-ready)
2 = PGRES_POLLING_WRITING    (wait for fd WRITE-ready)
3 = PGRES_POLLING_OK         (connected; stop polling)


465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
 
# File 'lib/tep/pg.rb', line 465

def self.async_connect(conninfo)
  h = Pg.tep_pg_connect_start(conninfo)
  if h < 0
    return -1
  end
  fd = Pg.tep_pg_socket(h)
  while true
    state = Pg.tep_pg_connect_poll(h)
    if state == 3
      # PGRES_POLLING_OK
      Pg.tep_pg_set_client_encoding(h, "UTF8")
      return h
    end
    if state == 0
      # PGRES_POLLING_FAILED. The shim has already stashed the
      # error message; we PQfinish the slot.
      Pg.tep_pg_finish(h)
      return -1
    end
    mode = state == 1 ? Tep::Scheduler::READ : Tep::Scheduler::WRITE
    Tep::Scheduler.io_wait(fd, mode, 10)
  end
  -1
end

.drain_remaining_results(pgh) ⇒ Object

After the first PQgetResult returned a real Result, libpq requires the conn be drained via additional PQgetResult calls until NULL is returned. This is a fast in-memory drain (no network), but it has to happen between async_exec calls or the next send_query will fail. Each tep_pg_get_result call that produces a non-NULL result stashes it in the slot table; we PQclear those immediately since they're trailing status results we don't expose.



552
553
554
555
556
557
558
559
560
561
562
563
 
# File 'lib/tep/pg.rb', line 552

def self.drain_remaining_results(pgh)
  while true
    rh = Pg.tep_pg_get_result(pgh)
    if rh < 0
      return 0
    end
    # A trailing result -- shouldn't normally happen for
    # single-statement queries, but defensively free.
    Pg.tep_pg_clear(rh)
  end
  0
end

.drain_send(pgh) ⇒ Object

Drain libpq's send buffer. PQflush returns 0 when done; 1 when the kernel send-buffer is full and we should park on WRITE-ready; -1 on error. Timeout is generous (10s); a genuinely-stuck PG is the rare case worth bailing on.



509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
 
# File 'lib/tep/pg.rb', line 509

def self.drain_send(pgh)
  fd = Pg.tep_pg_socket(pgh)
  while true
    rc = Pg.tep_pg_flush(pgh)
    if rc == 0
      return 0
    end
    if rc < 0
      return -1
    end
    # rc == 1: send buffer full, park on writability.
    Tep::Scheduler.io_wait(fd, Tep::Scheduler::WRITE, 10)
  end
  0
end

.escape_string(s) ⇒ Object

Class-method form -- ruby-pg allows escape_string and quote_ident without a live conn. We route through slot 0 which the shim treats as "no conn, fall back to standalone PQescapeString". Use the instance method when a conn is available -- it goes through PQescapeStringConn which is the standards-compliant path.



583
584
585
 
# File 'lib/tep/pg.rb', line 583

def self.escape_string(s)
  Pg.tep_pg_escape_string(0, s)
end

.quote_ident(s) ⇒ Object 



587
588
589
590
591
592
 
# File 'lib/tep/pg.rb', line 587

def self.quote_ident(s)
  # PQescapeIdentifier requires a conn; without one we fall
  # through to "" which is wrong but rare. Apps with a live
  # PG::Connection should use the instance method.
  Pg.tep_pg_escape_identifier(0, s)
end

.raise_send_failure(conn) ⇒ Object

PQsendQuery returned 0 (immediate failure -- conn already closed, send buffer error, etc.). Mirror the error onto the conn's last_* and raise, matching the exec error path (ruby-pg surfaces a send failure as PG::UnableToSend < PG::Error). No SQLSTATE is available pre-result, so this maps to the transport leaf rather than going through raise_for_sqlstate.

Raises:



498
499
500
501
502
503
 
# File 'lib/tep/pg.rb', line 498

def self.raise_send_failure(conn)
  conn.last_sqlstate = ""
  conn.last_error_message = conn.error_message
  conn.last_result_rh = -1
  raise PG::UnableToSend, conn.error_message
end

.record_error_if_any(conn, r) ⇒ Object

If the Result is in an error state, mirror SQLSTATE + message + result-handle onto the conn so post-rescue (or post-if !r.ok?) callers can read them via conn.last_*. No raise here -- see the docstring on exec for why.



598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
 
# File 'lib/tep/pg.rb', line 598

def self.record_error_if_any(conn, r)
  st = r.status
  if st == Pg::RES_TUPLES || st == Pg::RES_COMMAND || st == Pg::RES_EMPTY
    return 0
  end
  sqlstate = r.error_field(PG::DIAG_SQLSTATE)
  msg = r.error_message
  if msg.length == 0
    msg = conn.error_message
  end
  conn.last_sqlstate = sqlstate
  conn.last_error_message = msg
  # Free the failed PGresult NOW: once we raise out of
  # exec/exec_params the caller's `r.clear` never runs, so this is
  # the only chance to release it. The SQLSTATE / message are
  # already copied onto conn.last_* (Strings) for post-rescue
  # inspection, so dropping the handle loses nothing callers need.
  conn.last_result_rh = -1
  r.clear
  # ruby-pg / AR parity: raise the SQLSTATE-mapped PG::Error
  # subclass (live since matz/spinel#627 + #1041 -- namespaced
  # raise + hierarchy-walking rescue). Callers `rescue
  # PG::UniqueViolation` (leaf) or `rescue PG::Error` (base).
  PG.raise_for_sqlstate(sqlstate, msg)
  0
end

.wait_for_result_ready(pgh) ⇒ Object

Wait until PQisBusy returns 0 (PQgetResult won't block). Pumps PQconsumeInput in between io_wait calls so the libpq state machine advances. Timeout is generous (30s) since the query itself can take that long; the io_wait timeout is per-iteration, not cumulative.



530
531
532
533
534
535
536
537
538
539
540
541
542
 
# File 'lib/tep/pg.rb', line 530

def self.wait_for_result_ready(pgh)
  fd = Pg.tep_pg_socket(pgh)
  while true
    if Pg.tep_pg_consume_input(pgh) != 1
      return -1
    end
    if Pg.tep_pg_is_busy(pgh) == 0
      return 0
    end
    Tep::Scheduler.io_wait(fd, Tep::Scheduler::READ, 30)
  end
  0
end

Instance Method Details

#async_exec(sql) ⇒ Object

Explicit async exec. Same shape as exec but doesn't context-detect -- always uses the libpq async surface. If called outside Tep::Server::Scheduled, Tep::Scheduler.io_wait falls back to a single-shot poll(2), so this still works under prefork (just without the cross-fiber concurrency win).



391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
 
# File 'lib/tep/pg.rb', line 391

def async_exec(sql)
  Pg.tep_pg_set_nonblocking(@pgh, 1)
  ok = Pg.tep_pg_send_query(@pgh, sql)
  if ok != 1
    if @seed_conn
      return PG::Result.new(-1)
    end
    Connection.raise_send_failure(self)
  end
  Connection.drain_send(@pgh)
  Connection.wait_for_result_ready(@pgh)
  rh = Pg.tep_pg_get_result(@pgh)
  r = PG::Result.new(rh)
  # Drain trailing NULL terminator (libpq requires reading
  # until PQgetResult returns NULL to mark the conn ready for
  # the next send_query).
  Connection.drain_remaining_results(@pgh)
  Connection.record_error_if_any(self, r)
  r
end

#async_exec_params(sql, params) ⇒ Object

Parameterised async exec. params is an Array of String / Integer / nil; same conversion as exec_params.



414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
 
# File 'lib/tep/pg.rb', line 414

def async_exec_params(sql, params)
  Pg.tep_pg_param_clear
  i = 0
  n = params.length
  while i < n
    p = params[i]
    if p == nil
      Pg.tep_pg_param_push_null
    else
      Pg.tep_pg_param_push_str(p.to_s)
    end
    i += 1
  end
  async_exec_params_after_clear(sql)
end

#async_exec_params_after_clear(sql) ⇒ Object

Internal: param accumulator has already been populated by the caller (either exec_params routing here on context detect, or async_exec_params after its own push loop).



433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
 
# File 'lib/tep/pg.rb', line 433

def async_exec_params_after_clear(sql)
  Pg.tep_pg_set_nonblocking(@pgh, 1)
  ok = Pg.tep_pg_send_query_params(@pgh, sql)
  if ok != 1
    if @seed_conn
      return PG::Result.new(-1)
    end
    Connection.raise_send_failure(self)
  end
  Connection.drain_send(@pgh)
  Connection.wait_for_result_ready(@pgh)
  rh = Pg.tep_pg_get_result(@pgh)
  r = PG::Result.new(rh)
  Connection.drain_remaining_results(@pgh)
  Connection.record_error_if_any(self, r)
  r
end

#close ⇒ Object 



248
249
250
251
252
253
254
 
# File 'lib/tep/pg.rb', line 248

def close
  if @pgh >= 0
    Pg.tep_pg_finish(@pgh)
    @pgh = -1
  end
  0
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


244
245
246
 
# File 'lib/tep/pg.rb', line 244

def connected?
  @pgh >= 0
end

#error_message ⇒ Object 



279
280
281
 
# File 'lib/tep/pg.rb', line 279

def error_message
  @pgh < 0 ? "" : Pg.tep_pg_error_message(@pgh)
end

#escape_identifier(s) ⇒ Object 



569
570
571
 
# File 'lib/tep/pg.rb', line 569

def escape_identifier(s)
  Pg.tep_pg_escape_identifier(@pgh, s)
end

#escape_literal(s) ⇒ Object 



573
574
575
 
# File 'lib/tep/pg.rb', line 573

def escape_literal(s)
  Pg.tep_pg_escape_literal(@pgh, s)
end

#escape_string(s) ⇒ Object 



565
566
567
 
# File 'lib/tep/pg.rb', line 565

def escape_string(s)
  Pg.tep_pg_escape_string(@pgh, s)
end

#exec(sql) ⇒ Object

Run a no-params query. Returns a PG::Result on success.

ON ERROR IT RAISES the SQLSTATE-mapped PG::Error subclass (PG::UniqueViolation, PG::UndefinedTable, ... -> PG::ServerError for unmapped states) -- the ruby-pg / AR shape. The failed PGresult is freed before the raise; the SQLSTATE / message stay readable on conn.last_sqlstate / #last_error_message for post-rescue inspection:

begin
  c.exec(sql)
rescue PG::UniqueViolation => e
  ...                     # e.message + c.last_sqlstate
rescue PG::Error => e     # base catches any server error
  ...
end

Raising (instead of the old Result-on-error sentinel) became viable once spinel learned namespaced raise + hierarchy-walking rescue (matz/spinel#627 + #1041). NB: PG.connect is the one path that still does NOT raise -- it returns a connection-failed instance so PG::Pool can type-seed without a live server (check conn.connected?).

Under Tep::Server::Scheduled this routes through the libpq async surface (PQsendQuery + PQflush + PQconsumeInput parked on Tep::Scheduler.io_wait), so other fibers in the same worker can run while the query is in flight. Under prefork it routes through the blocking PQexec. Both raise identically.



350
351
352
353
354
355
356
357
358
 
# File 'lib/tep/pg.rb', line 350

def exec(sql)
  if Tep::Scheduler.scheduled_context?
    return async_exec(sql)
  end
  rh = Pg.tep_pg_exec(@pgh, sql)
  r = PG::Result.new(rh)
  Connection.record_error_if_any(self, r)
  r
end

#exec_params(sql, params) ⇒ Object

Parameterised query with positional binds ($1, $2, ...). params is an Array of String / Integer / nil. Same raise-on-error contract + auto-routing as exec.



363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
 
# File 'lib/tep/pg.rb', line 363

def exec_params(sql, params)
  Pg.tep_pg_param_clear
  i = 0
  n = params.length
  while i < n
    p = params[i]
    if p == nil
      Pg.tep_pg_param_push_null
    else
      Pg.tep_pg_param_push_str(p.to_s)
    end
    i += 1
  end
  if Tep::Scheduler.scheduled_context?
    return async_exec_params_after_clear(sql)
  end
  rh = Pg.tep_pg_exec_params(@pgh, sql)
  r = PG::Result.new(rh)
  Connection.record_error_if_any(self, r)
  r
end

#finish ⇒ Object 



256
257
258
 
# File 'lib/tep/pg.rb', line 256

def finish
  close
end

#last_notify_channel ⇒ Object 



313
314
315
 
# File 'lib/tep/pg.rb', line 313

def last_notify_channel
  Pg.tep_pg_notify_channel
end

#last_notify_payload ⇒ Object 



317
318
319
 
# File 'lib/tep/pg.rb', line 317

def last_notify_payload
  Pg.tep_pg_notify_payload
end

#listen(channel) ⇒ Object

LISTEN / NOTIFY (Battery 2 chunk 2.2). Used by Tep::Broadcast's PG backend for cross-worker pub/sub. Channel names must be safe SQL identifiers (no caller- controlled interpolation -- use a hard-coded constant). Payload max size is 8000 bytes per PG default.



288
289
290
291
 
# File 'lib/tep/pg.rb', line 288

def listen(channel)
  return -1 if @pgh < 0
  Pg.tep_pg_listen(@pgh, channel)
end

#notify(channel, payload) ⇒ Object 



298
299
300
301
 
# File 'lib/tep/pg.rb', line 298

def notify(channel, payload)
  return -1 if @pgh < 0
  Pg.tep_pg_notify(@pgh, channel, payload)
end

#poll_notification(timeout_ms) ⇒ Object

Block up to timeout_ms waiting for one notification on the connection. Returns 1 on receipt (caller then reads #last_notify_channel + #last_notify_payload), 0 on timeout, -1 on connection error. Connection must already be in LISTEN mode for the channel of interest.



308
309
310
311
 
# File 'lib/tep/pg.rb', line 308

def poll_notification(timeout_ms)
  return -1 if @pgh < 0
  Pg.tep_pg_poll_notification(@pgh, timeout_ms)
end

#reset ⇒ Object 



260
261
262
263
264
265
 
# File 'lib/tep/pg.rb', line 260

def reset
  if @pgh >= 0
    Pg.tep_pg_reset(@pgh)
  end
  self
end

#server_version ⇒ Object 



275
276
277
 
# File 'lib/tep/pg.rb', line 275

def server_version
  @pgh < 0 ? 0 : Pg.tep_pg_server_version(@pgh)
end

#status ⇒ Object 



267
268
269
 
# File 'lib/tep/pg.rb', line 267

def status
  @pgh < 0 ? PG::CONNECTION_BAD : Pg.tep_pg_status(@pgh)
end

#transaction_status ⇒ Object 



271
272
273
 
# File 'lib/tep/pg.rb', line 271

def transaction_status
  @pgh < 0 ? PG::PQTRANS_UNKNOWN : Pg.tep_pg_transaction_status(@pgh)
end

#unlisten(channel) ⇒ Object 



293
294
295
296
 
# File 'lib/tep/pg.rb', line 293

def unlisten(channel)
  return -1 if @pgh < 0
  Pg.tep_pg_unlisten(@pgh, channel)
end