Class: Smith::PersistenceAdapters::RedisStore

Inherits:
Object
  • Object
show all
Includes:
RedisClientAccess
Defined in:
lib/smith/persistence_adapters/redis_store.rb

Constant Summary collapse

TRANSIENT_ERROR_NAMES =

Redis transient errors — narrow list; non-transient errors (CommandError, etc.) propagate up immediately. Pattern matches Redis::BaseConnectionError if loaded (covers Connection/Timeout) via class-name guard so Smith doesn't require redis at load time.

%w[
  Redis::BaseConnectionError
  Redis::TimeoutError
  Redis::CannotConnectError
  Redis::ConnectionError
].freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis:, namespace: "smith", identity: nil) ⇒ RedisStore

Returns a new instance of RedisStore.



33
34
35
36
37
38
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 33

def initialize(redis:, namespace: "smith", identity: nil)
  @redis_source = redis
  @namespace = namespace.nil? ? nil : namespace.to_s.dup.freeze
  @persistence_identity = identity.to_s.dup.freeze if identity
  @client_resolution_mutex = Mutex.new
end

Instance Attribute Details

#persistence_identity ⇒ Object (readonly)

Returns the value of attribute persistence_identity.



31
32
33
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 31

def persistence_identity
  @persistence_identity
end

Class Method Details

.transient_errors ⇒ Object 



23
24
25
26
27
28
29
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 23

def self.transient_errors
  TRANSIENT_ERROR_NAMES.filter_map do |name|
    Object.const_get(name)
  rescue NameError
    nil
  end + [Errno::ECONNREFUSED, Errno::ETIMEDOUT, Errno::EPIPE]
end

Instance Method Details

#delete(key) ⇒ Object 



56
57
58
59
60
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 56

def delete(key)
  Retry.with_retries(operation: :delete, transient: self.class.transient_errors) do
    client.del(namespaced(key), namespaced_heartbeat(key))
  end
end

#fetch(key) ⇒ Object 



50
51
52
53
54
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 50

def fetch(key)
  Retry.with_retries(operation: :fetch, transient: self.class.transient_errors) do
    client.get(namespaced(key))
  end
end

#last_heartbeat(key) ⇒ Object 



76
77
78
79
80
81
82
83
84
85
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 76

def last_heartbeat(key)
  Retry.with_retries(operation: :last_heartbeat, transient: self.class.transient_errors) do
    raw = client.get(namespaced_heartbeat(key))
    next nil if raw.nil?

    Time.parse(raw).utc
  rescue ArgumentError
    nil
  end
end

#record_heartbeat(key, ttl: Smith.config.persistence_ttl) ⇒ Object 



65
66
67
68
69
70
71
72
73
74
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 65

def record_heartbeat(key, ttl: Smith.config.persistence_ttl)
  Retry.with_retries(operation: :record_heartbeat, transient: self.class.transient_errors) do
    iso = Time.now.utc.iso8601
    if ttl
      client.set(namespaced_heartbeat(key), iso, ex: ttl)
    else
      client.set(namespaced_heartbeat(key), iso)
    end
  end
end

#replace_exact(key, payload, expected_payload:, ttl: Smith.config.persistence_ttl) ⇒ Object 



105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 105

def replace_exact(key, payload, expected_payload:, ttl: Smith.config.persistence_ttl)
  without_reconnection do |redis|
    RedisExactWrite.new(
      client: redis,
      key: key,
      storage_key: namespaced(key),
      payload: payload,
      expected_payload: expected_payload,
      ttl: ttl
    ).call
  end
rescue *self.class.transient_errors => e
  raise Smith::PersistenceIOError.new(operation: :replace_exact, cause: e)
end

#store(key, payload, ttl: Smith.config.persistence_ttl) ⇒ Object 



40
41
42
43
44
45
46
47
48
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 40

def store(key, payload, ttl: Smith.config.persistence_ttl)
  Retry.with_retries(operation: :store, transient: self.class.transient_errors) do
    if ttl
      client.set(namespaced(key), payload, ex: ttl)
    else
      client.set(namespaced(key), payload)
    end
  end
end

#store_versioned(key, payload, expected_version:, ttl: Smith.config.persistence_ttl) ⇒ Object

Optimistic locking via Redis WATCH/MULTI/EXEC. Raises Smith::PersistenceVersionConflict on a stale expected_version OR on EXEC failure (WATCH detected concurrent write).



90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 90

def store_versioned(key, payload, expected_version:, ttl: Smith.config.persistence_ttl)
  without_reconnection do |redis|
    RedisVersionedWrite.new(
      client: redis,
      key: key,
      storage_key: namespaced(key),
      payload: payload,
      expected_version: expected_version,
      ttl: ttl
    ).call
  end
rescue *self.class.transient_errors => e
  raise Smith::PersistenceIOError.new(operation: :store_versioned, cause: e)
end

#transaction_identity ⇒ Object 



63
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 63

def transaction_identity = nil

#transaction_open? ⇒ Boolean

Returns:

  • (Boolean) 


62
 
# File 'lib/smith/persistence_adapters/redis_store.rb', line 62

def transaction_open? = false