Class: OpenReceive::Server::Swap::TransientSwapCache

Inherits:
Object
  • Object
show all
Defined in:
lib/openreceive/server/swap/transient_cache.rb

Overview

Ruby port of packages/js/node/src/swap/limits-cache.ts: a disposable, process-local provider catalog/rate cache. It has no storage adapter.

Like the Node in-flight promise map, concurrent resolves of the same key join one fetch: the first caller claims the key under the monitor, runs fetch outside it, and writes back under the monitor; joiners wait on a per-key condition. Different keys never block each other.

Constant Summary collapse

MAX_STALE_SECONDS = 
48 * 60 * 60
REFRESH_CLAIM_SECONDS = 
60

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(clock, warn: nil) ⇒ TransientSwapCache

Returns a new instance of TransientSwapCache.



23
24
25
26
27
28
29
 
# File 'lib/openreceive/server/swap/transient_cache.rb', line 23

def initialize(clock, warn: nil)
  @clock = clock
  @warn = warn
  @states = {}
  @inflight = {}
  @monitor = Monitor.new
end

Class Method Details

.limits_meta_key(provider_name) ⇒ Object 



19
20
21
 
# File 'lib/openreceive/server/swap/transient_cache.rb', line 19

def self.limits_meta_key(provider_name)
  "swap_limits:#{provider_name}"
end

Instance Method Details

#resolve(key, refresh_seconds:, max_stale_seconds:, fetch:, serialize:, deserialize:, claim_seconds: REFRESH_CLAIM_SECONDS, serve_stale_on_failure: true) ⇒ Object 



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/openreceive/server/swap/transient_cache.rb', line 31

def resolve(key, refresh_seconds:, max_stale_seconds:, fetch:, serialize:, deserialize:,
            claim_seconds: REFRESH_CLAIM_SECONDS, serve_stale_on_failure: true)
  now = nil
  state = nil
  claim = nil
  @monitor.synchronize do
    now = @clock.call
    state = @states[key]
    if state && state[:value] && state[:fetched_at] && now - state[:fetched_at] < refresh_seconds
      return deserialize.call(state[:value])
    end
    if state && state[:failed_at] && now - state[:failed_at] < claim_seconds
      return stale_or_raise(key, state, now, max_stale_seconds, serve_stale_on_failure, deserialize)
    end

    active = @inflight[key]
    if active
      active[:cond].wait_until { active[:settled] }
      raise active[:error] unless active[:error].nil?

      return active[:value]
    end
    claim = { settled: false, cond: @monitor.new_cond }
    @inflight[key] = claim
  end

  begin
    result = begin
      value = fetch.call
      @monitor.synchronize { @states[key] = { value: serialize.call(value), fetched_at: now } }
      value
    rescue StandardError => e
      failed = {
        failed_at: now,
        error: e.message
      }
      failed[:value] = state[:value] if state && state[:value]
      failed[:fetched_at] = state[:fetched_at] if state && state[:fetched_at]
      @monitor.synchronize { @states[key] = failed }
      stale_or_raise(key, failed, now, max_stale_seconds, serve_stale_on_failure, deserialize, cause: e)
    end
    settle(key, claim, value: result)
    result
  rescue StandardError => e
    settle(key, claim, error: e)
    raise
  end
end