Class: Waitmate::Store::SolidCache

Inherits:
Object
  • Object
show all
Defined in:
lib/waitmate/store/solid_cache.rb

Overview

SQL-backed fallback adapter built on top of SolidCache::Entry.

Solid Cache is a fixed-schema key/value store, so Waitmate queue state is separated into two key prefixes:

waitmate:waiting:{queue}:{identity}  — waiting entries
waitmate:active:{queue}:{identity}   — admitted entries

This mirrors the Redis adapter's separate sorted sets and lets SQL LIKE narrow scans to one state before Ruby-side JSON parsing. Active scans are bounded by max_concurrent; waiting scans are bounded by queue depth ahead of the target entry.

FIFO ordering relies on an enqueued_at timestamp stored in the JSON value, avoiding reliance on the binary created_at column which has cross-Rails-version comparison issues.

Single-key lookups use SolidCache::Entry.read / delete_by_key which operate on the indexed key_hash integer column, avoiding binary key column comparison entirely.

Admission is serialized per queue with a mutex row so capacity accounting never follows a read-modify-write path under concurrency.

Constant Summary collapse

KEY_PREFIX = 
"waitmate"

Instance Method Summary collapse

Constructor Details

#initialize(config: Waitmate.configuration) ⇒ SolidCache

Returns a new instance of SolidCache.



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/waitmate/store/solid_cache.rb', line 38

def initialize(config: Waitmate.configuration)
  begin
    require "solid_cache" unless defined?(::SolidCache)
  rescue LoadError
    # Fall through to the contract check below.
  end

  unless defined?(::SolidCache::Entry)
    raise ConfigurationError,
      "Solid Cache gem is not available. Add `gem 'solid_cache'` to your Gemfile " \
      "to use the Solid Cache adapter."
  end

  @queue_ttl = config.queue_ttl
end

Instance Method Details

#active_count(queue_name) ⇒ Object

active_count(queue_name) -> Integer



120
121
122
123
124
125
126
 
# File 'lib/waitmate/store/solid_cache.rb', line 120

def active_count(queue_name)
  now = current_timestamp

  with_ar do
    active_count_internal(queue_name, now)
  end
end

#admit(queue_name, max_concurrent, count: max_concurrent) ⇒ Object

admit(queue_name, max_concurrent, count: max_concurrent) -> Array



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/waitmate/store/solid_cache.rb', line 129

def admit(queue_name, max_concurrent, count: max_concurrent)
  now = current_timestamp
  active_ttl = @queue_ttl

  with_ar do
    admitted = []

    ActiveRecord::Base.transaction do
      ensure_mutex(queue_name)
      expire_stale_internal(queue_name, now)

      active = active_count_internal(queue_name, now)
      available = max_concurrent - active
      break admitted if available <= 0

      admit_count = [available, count].min

      # Separate waiting prefix guarantees all returned rows are waiting.
      # No Ruby-side state filter needed — fixes the S2-NEWBUG capacity bug.
      # Sort by enqueued_at from the JSON value for FIFO ordering.
      # uncached is required because SolidCache's own write/delete methods do
      # not dirty the Rails query cache, so a subsequent scan in the same
      # request (e.g. release → admit) could read stale waiting rows.
      waiting = ::SolidCache::Entry.uncached do
        ::SolidCache::Entry
          .where("key LIKE ?", waiting_key_pattern(queue_name))
          .to_a
          .map { |entry| [entry, parse_value(entry.value, key: entry.key)] }
          .select { |_, value| value["expires_at"] > now }
          .sort_by { |_, value| value["enqueued_at"] }
          .first(admit_count)
      end

      waiting.each do |entry, value|
        identity = identity_from_key(entry.key, queue_name)
        value["state"] = "active"
        value["expires_at"] = now + active_ttl
        ::SolidCache::Entry.delete_by_key(waiting_entry_key(queue_name, identity))
        ::SolidCache::Entry.write(active_entry_key(queue_name, identity), value.to_json)
        admitted << identity
      end
    end

    admitted
  end
end

#enqueue(queue_name, identity, ttl: nil) ⇒ Object

enqueue(queue_name, identity, ttl: nil) -> Integer (1-based position) or 0 (already active)



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
 
# File 'lib/waitmate/store/solid_cache.rb', line 55

def enqueue(queue_name, identity, ttl: nil)
  ttl ||= @queue_ttl
  now = current_timestamp
  expires_at = now + ttl
  a_key = active_entry_key(queue_name, identity)
  w_key = waiting_entry_key(queue_name, identity)

  with_ar do
    # Already active and not expired?
    active_value = read_value(a_key)
    if active_value
      return 0 if active_value["expires_at"] > now
    end

    # Already waiting and not expired?
    waiting_value = read_value(w_key)
    if waiting_value
      if waiting_value["expires_at"] > now
        waiting_value["expires_at"] = expires_at
        ::SolidCache::Entry.write(w_key, waiting_value.to_json)
        return waiting_position(queue_name, waiting_value["enqueued_at"], now)
      end
      # Expired waiting entry — delete so the new write gets a fresh enqueued_at
      ::SolidCache::Entry.delete_by_key(w_key)
    end

    # New entry
    value = {state: "waiting", expires_at: expires_at, enqueued_at: now}
    ::SolidCache::Entry.write(w_key, value.to_json)
    waiting_position(queue_name, now, now)
  end
end

#expire_stale(queue_name) ⇒ Object

expire_stale(queue_name) -> Hash Integer, active: Integer



217
218
219
220
221
222
223
224
225
226
 
# File 'lib/waitmate/store/solid_cache.rb', line 217

def expire_stale(queue_name)
  now = current_timestamp

  with_ar do
    ActiveRecord::Base.transaction do
      ensure_mutex(queue_name)
      expire_stale_internal(queue_name, now)
    end
  end
end

#heartbeat(queue_name, identity, ttl: nil) ⇒ Object

heartbeat(queue_name, identity, ttl: nil) -> Boolean



187
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
 
# File 'lib/waitmate/store/solid_cache.rb', line 187

def heartbeat(queue_name, identity, ttl: nil)
  ttl ||= @queue_ttl
  now = current_timestamp
  w_key = waiting_entry_key(queue_name, identity)
  a_key = active_entry_key(queue_name, identity)

  with_ar do
    # Check waiting first
    value = read_value(w_key)
    if value
      return false if value["expires_at"] <= now
      value["expires_at"] = now + ttl
      ::SolidCache::Entry.write(w_key, value.to_json)
      return true
    end

    # Check active
    value = read_value(a_key)
    if value
      return false if value["expires_at"] <= now
      value["expires_at"] = now + ttl
      ::SolidCache::Entry.write(a_key, value.to_json)
      return true
    end

    false
  end
end

#position(queue_name, identity) ⇒ Object

position(queue_name, identity) -> Integer (1-based), 0 (active), or nil



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
 
# File 'lib/waitmate/store/solid_cache.rb', line 89

def position(queue_name, identity)
  now = current_timestamp
  a_key = active_entry_key(queue_name, identity)
  w_key = waiting_entry_key(queue_name, identity)

  with_ar do
    # Check active
    active_value = read_value(a_key)
    if active_value
      if active_value["expires_at"] <= now
        ::SolidCache::Entry.delete_by_key(a_key)
        return nil
      end
      return 0
    end

    # Check waiting
    waiting_value = read_value(w_key)
    if waiting_value
      if waiting_value["expires_at"] <= now
        ::SolidCache::Entry.delete_by_key(w_key)
        return nil
      end
      return waiting_position(queue_name, waiting_value["enqueued_at"], now)
    end

    nil
  end
end

#release(queue_name, identity) ⇒ Object

release(queue_name, identity) -> Boolean



177
178
179
180
181
182
183
184
 
# File 'lib/waitmate/store/solid_cache.rb', line 177

def release(queue_name, identity)
  key = active_entry_key(queue_name, identity)

  with_ar do
    result = ::SolidCache::Entry.delete_by_key(key)
    result.is_a?(Integer) ? result > 0 : result
  end
end