Class: Pgtk::Stash

Inherits:
Object
  • Object
show all
Defined in:
lib/pgtk/stash.rb

Overview

Database query cache implementation.

Provides a caching layer for PostgreSQL queries, automatically invalidating the cache when tables are modified. Read queries are cached while write queries bypass the cache and invalidate related cached entries.

Tables that are too write-heavy to benefit from caching can be listed via the volatile: constructor parameter; any read whose FROM/JOIN set touches one of them bypasses the cache entirely.

Thread-safe with read-write locking.

The implementation is very naive! Use it at your own risk.

Author

Yegor Bugayenko (yegor256@gmail.com)

Copyright

Copyright (c) 2019-2026 Yegor Bugayenko

License

MIT

Examples:

Basic usage

pool = Pgtk::Pool.new(...)
stash = Pgtk::Stash.new(pool, cap: 1000, refill: 30)
stash.start!
result = stash.exec('SELECT * FROM users WHERE id = $1', [42])

Instance Method Summary collapse

Constructor Details

#initialize(pool, stash: { queries: {}, tables: {}, table_mod: {}, table_inflight: {} }, loog: Loog::NULL, entrance: Concurrent::ReentrantReadWriteLock.new, volatile: [], refill: 16, delay: 0, maxqueue: 128, threads: 4, cap: 10_000, capping: 60, retire: 15 * 60, retirement: 60) ⇒ Stash

Initialize a new Stash with query caching.

Set any of the intervals to nil to disable the cron.

Parameters:

  • pool (Object) —

    The underlying connection pool

  • stash (Hash) (defaults to: { queries: {}, tables: {}, table_mod: {}, table_inflight: {} }) —

    Internal cache structure

  • refill (Float) (defaults to: 16) —

    Interval in seconds between background refill tasks

  • delay (Float) (defaults to: 0) —

    A pause in seconds before making a refill

  • maxqueue (Integer) (defaults to: 128) —

    Maximum number of refilling tasks in the thread pool queue

  • threads (Integer) (defaults to: 4) —

    Number of worker threads for cache refilling

  • cap (Integer) (defaults to: 10_000) —

    Maximum number of cached query results to retain

  • capping (Float) (defaults to: 60) —

    Interval in seconds between cache cap enforcement tasks

  • retire (Integer) (defaults to: 15 * 60) —

    Maximum age in seconds to keep a query in cache

  • retirement (Float) (defaults to: 60) —

    Interval in seconds between retirement tasks

  • loog (Loog) (defaults to: Loog::NULL) —

    Logger instance

  • entrance (Concurrent::ReentrantReadWriteLock) (defaults to: Concurrent::ReentrantReadWriteLock.new) —

    Read-write lock for thread-safe access

  • volatile (Array<String>) (defaults to: []) —

    Table names that must never be cached



76
77
78
79
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
 
# File 'lib/pgtk/stash.rb', line 76

def initialize(
  pool,
  stash: { queries: {}, tables: {}, table_mod: {}, table_inflight: {} },
  loog: Loog::NULL,
  entrance: Concurrent::ReentrantReadWriteLock.new,
  volatile: [],
  refill: 16,
  delay: 0,
  maxqueue: 128,
  threads: 4,
  cap: 10_000,
  capping: 60,
  retire: 15 * 60,
  retirement: 60
)
  @pool = pool
  @stash = stash
  @stash[:table_mod] ||= {}
  unless @stash[:table_inflight].default_proc
    @stash[:table_inflight] = Hash.new { |h, k| h[k] = Concurrent::AtomicFixnum.new(0) }
  end
  @loog = loog
  @entrance = entrance
  @volatile = Array(volatile).map(&:to_s).freeze
  @refill = refill
  @delay = delay
  @maxqueue = maxqueue
  @threads = threads
  @cap = cap
  @capping = capping
  @retire = retire
  @retirement = retirement
end

Instance Method Details

#dump ⇒ String

Convert internal state into text.

Returns:

  • (String) —

    Multi-line text representation of the current cache state



128
129
130
131
132
 
# File 'lib/pgtk/stash.rb', line 128

def dump
  @entrance.with_read_lock do
    body(queries)
  end
end

#exec(query, params = [], result = 0) ⇒ PG::Result

Execute a SQL query with optional caching.

Parameters:

  • query (String, Array<String>) —

    The SQL query

  • params (Array) (defaults to: []) —

    Query parameters

  • result (Integer) (defaults to: 0) —

    Result format code

Returns:

  • (PG::Result) —

    Query result object



140
141
142
143
144
145
146
147
148
149
 
# File 'lib/pgtk/stash.rb', line 140

def exec(query, params = [], result = 0)
  pure = (query.is_a?(Array) ? query.join(' ') : query).gsub(/\s+/, ' ').strip
  if MODS_RE.match?(pure) || (WITH_RE.match?(pure) && ALTS_RE.match?(pure))
    modify(pure, params, result)
  elsif /(^|\s)pg_[a-z_]+\(/.match?(pure) || (!@volatile.empty? && @volatile.intersect?(pure.scan(READS_RE).flatten))
    @pool.exec(pure, params, result)
  else
    select(pure, params, result)
  end
end

#session {|Pgtk::Stash| ... } ⇒ Object

Run statements on a single connection, without a transaction.

Yields:

Returns:

  • (Object) —

    The result of the block



165
166
167
168
169
 
# File 'lib/pgtk/stash.rb', line 165

def session
  @pool.session do |t|
    yield(Pgtk::Stash.new(t, stash: @stash, loog: @loog, entrance: @entrance, volatile: @volatile))
  end
end

#start! ⇒ void

This method returns an undefined value.

Start the connection pool and launch background cache management tasks.



113
114
115
116
117
 
# File 'lib/pgtk/stash.rb', line 113

def start!
  @pool.start!
  cascade!
  launch!
end

#transaction {|Pgtk::Stash| ... } ⇒ Object

Execute a database transaction.

Yields:

  • (Pgtk::Stash) —

    A stash connected to the transaction

Returns:

  • (Object) —

    The result of the block



155
156
157
158
159
 
# File 'lib/pgtk/stash.rb', line 155

def transaction
  @pool.transaction do |t|
    yield(Pgtk::Stash.new(t, stash: @stash, loog: @loog, entrance: @entrance, volatile: @volatile))
  end
end

#version ⇒ String

Get the PostgreSQL server version.

Returns:

  • (String) —

    Version string of the database server



121
122
123
 
# File 'lib/pgtk/stash.rb', line 121

def version
  @pool.version
end