Class: Daytona::EventSubscriptionManager

Inherits:
Object
  • Object
show all
Defined in:
lib/daytona/common/event_subscription_manager.rb

Overview

Tracks dispatcher subscriptions with TTL auto-expiry.

All subscriptions share ONE lazily started expiry worker thread instead of a dedicated sleeping thread per subscription, so listing many sandboxes costs one thread total, not one thread each (https://github.com/daytona/clients/issues/108). #refresh only bumps the subscription deadline — it never creates or kills threads.

Instance Method Summary collapse

Constructor Details

#initialize(dispatcher = nil, ttl_seconds: SUBSCRIPTION_TTL) ⇒ EventSubscriptionManager

Returns a new instance of EventSubscriptionManager.



20
21
22
23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 20

def initialize(dispatcher = nil, ttl_seconds: SUBSCRIPTION_TTL)
  @dispatcher = dispatcher
  @ttl = ttl_seconds
  @subscriptions = {}
  # Deadline-ordered [expires_at, sub_id] entries. Entries are lazily
  # invalidated: a popped entry is discarded when its subscription is gone,
  # or re-queued at the current deadline when it was refreshed meanwhile.
  # Invariant: every live subscription has at least one queue entry.
  @expiry_queue = []
  @mutex = Mutex.new
  @cond = ConditionVariable.new
  @worker = nil
  @closed = false
end

Instance Method Details

#refresh(sub_id) ⇒ Object 



66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 66

def refresh(sub_id)
  @mutex.synchronize do
    return false if @closed

    subscription = @subscriptions[sub_id]
    return false unless subscription

    # No queue entry or worker wake-up needed: when the old deadline pops,
    # the worker sees the newer expires_at and re-queues the subscription.
    subscription[:expires_at] = monotonic_now + @ttl
    true
  end
end

#shutdown ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 95

def shutdown
  subscriptions = nil

  @mutex.synchronize do
    @closed = true
    subscriptions = @subscriptions.values
    @subscriptions = {}
    @expiry_queue.clear
    @cond.broadcast
  end

  subscriptions.each { |subscription| subscription[:unsubscribe].call }
end

#subscribe(resource_id:, handler:, events:) ⇒ Object 



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
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 35

def subscribe(resource_id:, handler:, events:)
  @mutex.synchronize do
    return nil if @closed || @dispatcher.nil?
  end

  unsubscribe = @dispatcher.subscribe(resource_id, events:, &handler)
  sub_id = SecureRandom.hex(16)
  rollback_unsubscribe = nil

  @mutex.synchronize do
    if @closed
      # Rollback dispatcher subscription on failure
      rollback_unsubscribe = unsubscribe
      next
    end

    expires_at = monotonic_now + @ttl
    @subscriptions[sub_id] = { unsubscribe:, expires_at: }
    push_entry_locked(expires_at, sub_id)
    ensure_worker_locked
    @cond.signal
  end

  if rollback_unsubscribe
    rollback_unsubscribe.call
    return nil
  end

  sub_id
end

#unsubscribe(sub_id) ⇒ Object 



80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 80

def unsubscribe(sub_id)
  subscription = @mutex.synchronize do
    removed = @subscriptions.delete(sub_id)
    # The stale queue entry is discarded when the worker pops it — except
    # when it was the last subscription: drop the leftovers and wake the
    # worker so it exits now instead of idling until the old deadline.
    if removed && @subscriptions.empty?
      @expiry_queue.clear
      @cond.signal
    end
    removed
  end
  subscription&.dig(:unsubscribe)&.call
end