Class: Langfuse::RailsCacheAdapter

Inherits:
Object
  • Object
show all
Includes:
StaleWhileRevalidate
Defined in:
lib/langfuse/rails_cache_adapter.rb

Overview

Rails.cache adapter for distributed caching with Redis

Wraps Rails.cache to provide distributed caching for prompts across multiple processes and servers. Requires Rails with Redis cache store.

rubocop:disable Metrics/ClassLength

Examples:

adapter = Langfuse::RailsCacheAdapter.new(ttl: 60)
adapter.set("greeting:1", prompt_data)
adapter.get("greeting:1") # => prompt_data

Constant Summary collapse

GENERATION_MEMO_TTL_SECONDS = 
1.0

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from StaleWhileRevalidate

#fetch_with_stale_while_revalidate, #initialize_swr, #refresh_async, #shutdown, #swr_enabled?, #write_with_stale_while_revalidate

Constructor Details

#initialize(ttl: 60, namespace: "langfuse", lock_timeout: 10, stale_ttl: 0, refresh_threads: 5, logger: default_logger) ⇒ RailsCacheAdapter

Initialize a new Rails.cache adapter

Parameters:

  • ttl (Integer) (defaults to: 60) —

    Time-to-live in seconds (default: 60)

  • namespace (String) (defaults to: "langfuse") —

    Cache key namespace (default: "langfuse")

  • lock_timeout (Integer) (defaults to: 10) —

    Lock timeout in seconds for stampede protection (default: 10)

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

    Stale TTL for SWR in seconds (default: 0, SWR disabled). Note: :indefinite is normalized to 1000 years by Config before being passed here.

  • refresh_threads (Integer) (defaults to: 5) —

    Number of background refresh threads (default: 5)

  • logger (Logger, nil) (defaults to: default_logger) —

    Logger instance for error reporting (default: nil, creates new logger)

Raises:



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 60

def initialize(ttl: 60, namespace: "langfuse", lock_timeout: 10, stale_ttl: 0, refresh_threads: 5,
               logger: default_logger)
  validate_rails_cache!

  @ttl = ttl
  @namespace = namespace
  @namespace_prefix = "#{namespace}:"
  @lock_timeout = lock_timeout
  @stale_ttl = stale_ttl
  @logger = logger
  @generation_memo = {}
  @generation_memo_mutex = Mutex.new
  initialize_swr(refresh_threads: refresh_threads) if swr_enabled?
  ForkSafety.register(self)
end

Instance Attribute Details

#lock_timeout ⇒ Integer (readonly)

Returns Lock timeout in seconds for stampede protection.

Returns:

  • (Integer) —

    Lock timeout in seconds for stampede protection



39
40
41
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 39

def lock_timeout
  @lock_timeout
end

#logger ⇒ Logger (readonly)

Returns Logger instance for error reporting.

Returns:

  • (Logger) —

    Logger instance for error reporting



48
49
50
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 48

def logger
  @logger
end

#namespace ⇒ String (readonly)

Returns Cache key namespace.

Returns:

  • (String) —

    Cache key namespace



36
37
38
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 36

def namespace
  @namespace
end

#stale_ttl ⇒ Integer (readonly)

Returns Stale TTL for SWR in seconds.

Returns:

  • (Integer) —

    Stale TTL for SWR in seconds



42
43
44
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 42

def stale_ttl
  @stale_ttl
end

#thread_pool ⇒ Concurrent::CachedThreadPool? (readonly)

Returns Thread pool for background refreshes.

Returns:

  • (Concurrent::CachedThreadPool, nil) —

    Thread pool for background refreshes



45
46
47
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 45

def thread_pool
  @thread_pool
end

#ttl ⇒ Integer (readonly)

Returns Time-to-live in seconds.

Returns:

  • (Integer) —

    Time-to-live in seconds



33
34
35
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 33

def ttl
  @ttl
end

Class Method Details

.available? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Check whether Rails has a configured cache store.

Returns:

  • (Boolean) 


28
29
30
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 28

def self.available?
  !!(defined?(Rails) && Rails.respond_to?(:cache) && !Rails.cache.nil?)
end

.build_key(name, version: nil, label: nil) ⇒ String

Build a cache key from prompt name and options

Parameters:

  • name (String) —

    Prompt name

  • version (Integer, nil) (defaults to: nil) —

    Optional version

  • label (String, nil) (defaults to: nil) —

    Optional label

Returns:

  • (String) —

    Cache key



210
211
212
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 210

def self.build_key(name, version: nil, label: nil)
  PromptCache.build_key(name, version: version, label: label)
end

Instance Method Details

#clear ⇒ void

This method returns an undefined value.

Clear the entire Langfuse cache namespace

Note: This uses delete_matched which may not be available on all cache stores. Works with Redis, Memcached, and memory stores. File store support varies.



121
122
123
124
125
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 121

def clear
  # Delete all keys matching the namespace pattern
  Rails.cache.delete_matched("#{namespace}:*")
  clear_generation_memo
end

#clear_logically ⇒ Integer

Logically invalidate every generated storage key.

Returns:

  • (Integer) —

    New global generation



130
131
132
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 130

def clear_logically
  bump_generation(global_generation_key)
end

#delete(key) ⇒ Boolean

Delete one generated storage key.

Parameters:

  • key (String) —

    Cache key

Returns:

  • (Boolean) —

    true if an entry was removed



111
112
113
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 111

def delete(key)
  Rails.cache.delete(namespaced_key(key))
end

#empty? ⇒ Boolean

Check if cache is empty

Note: Rails.cache doesn't provide an efficient way to check if empty, so we return false to indicate this operation is not supported.

Returns:

  • (Boolean) —

    Always returns false (unsupported operation)



189
190
191
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 189

def empty?
  false
end

#entry(key) ⇒ Object?

Read a raw cache entry, including stale entries.

Parameters:

  • key (String) —

    Cache key

Returns:

  • (Object, nil) —

    Raw cache entry



88
89
90
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 88

def entry(key)
  Rails.cache.read(namespaced_key(key))
end

#fetch_with_lock(key, ttl: nil) { ... } ⇒ Object

Fetch a value from cache with lock for stampede protection

This method prevents cache stampedes (thundering herd) by ensuring only one process/thread fetches from the source when the cache is empty. Others wait for the first one to populate the cache.

Uses exponential backoff: 50ms, 100ms, 200ms (3 retries max, ~350ms total). If cache is still empty after waiting, falls back to fetching from source.

Examples:

cache.fetch_with_lock("greeting:v1") do
  api_client.get_prompt("greeting")
end

Parameters:

  • key (String) —

    Cache key

Yields:

  • Block to execute if cache miss (should fetch fresh data)

Returns:

  • (Object) —

    Cached or freshly fetched value



231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 231

def fetch_with_lock(key, ttl: nil)
  # 1. Check cache first (fast path - no lock needed)
  cached = get(key)
  return cached if cached

  # 2. Cache miss - try to acquire lock
  lock_key = build_lock_key(key)

  if acquire_lock(lock_key)
    begin
      # We got the lock - fetch from source and populate cache
      value = yield
      set(key, value, ttl: ttl)
      value
    ensure
      # Always release lock, even if block raises
      release_lock(lock_key)
    end
  else
    # Someone else has the lock - wait for them to populate cache
    cached = wait_for_cache(key)
    return cached if cached

    # Cache still empty after waiting - fall back to fetching ourselves
    # (This handles cases where lock holder crashed or took too long)
    yield
  end
end

#get(key) ⇒ Object?

Get a value from the cache

Parameters:

  • key (String) —

    Cache key

Returns:

  • (Object, nil) —

    Cached value or nil if not found/expired



80
81
82
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 80

def get(key)
  Rails.cache.read(namespaced_key(key))
end

#invalidate_name(name) ⇒ Integer

Logically invalidate every cache variant for one prompt name.

Parameters:

  • name (String) —

    Prompt name

Returns:

  • (Integer) —

    New name generation



138
139
140
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 138

def invalidate_name(name)
  bump_generation(name_generation_key(name))
end

#set(key, value, ttl: nil, stale_ttl: nil) ⇒ Object

Set a value in the cache

Parameters:

  • key (String) —

    Cache key

  • value (Object) —

    Value to cache

Returns:

  • (Object) —

    The cached value



97
98
99
100
101
102
103
104
105
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 97

def set(key, value, ttl: nil, stale_ttl: nil)
  # Total ttl when SWR is enabled, otherwise just ttl. Inlined (not pushed
  # to a shared helper) to keep this hot write path allocation-free.
  effective_ttl = ttl.nil? ? self.ttl : ttl
  effective_stale_ttl = stale_ttl.nil? ? self.stale_ttl : stale_ttl
  expires_in = swr_enabled? ? effective_ttl + effective_stale_ttl : effective_ttl
  Rails.cache.write(namespaced_key(key), value, expires_in:)
  value
end

#size ⇒ nil

Get current cache size

Note: Rails.cache doesn't provide a size method, so we return nil to indicate this operation is not supported.

Returns:

  • (nil) 


163
164
165
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 163

def size
  nil
end

#stats ⇒ Hash

Returns Prompt cache statistics.

Returns:

  • (Hash) —

    Prompt cache statistics



168
169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 168

def stats
  {
    backend: CacheBackend::RAILS,
    enabled: true,
    current_generation_entries: nil,
    orphaned_entries: nil,
    total_entries: nil,
    ttl: ttl,
    size: size,
    max_size: nil,
    global_generation: safe_generation_value(global_generation_key),
    unsupported_counts: CacheBackend::UNSUPPORTED_COUNT_KEYS
  }
end

#storage_key(logical_key, name:) ⇒ String

Build a generated storage key for the current cache generation.

Parameters:

  • logical_key (String) —

    Stable logical cache identity

  • name (String) —

    Prompt name

Returns:

  • (String) —

    Generated storage key



147
148
149
150
151
152
153
154
155
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 147

def storage_key(logical_key, name:)
  generated = PromptCache.storage_key(
    logical_key,
    name: name,
    global_generation: generation_value(global_generation_key),
    name_generation: generation_value(name_generation_key(name))
  )
  namespaced_key(generated)
end

#validate! ⇒ Boolean

Validate that Rails.cache is available for prompt caching.

rubocop:disable Naming/PredicateMethod

Returns:

  • (Boolean)

Raises:



198
199
200
201
 
# File 'lib/langfuse/rails_cache_adapter.rb', line 198

def validate!
  validate_rails_cache!
  true
end