Module: Models::AssistantConversationProcessingLockable

Extended by:
ActiveSupport::Concern
Included in:
AssistantConversation
Defined in:
app/concerns/models/assistant_conversation_processing_lockable.rb

Overview

Column-based processing lock for assistant conversations.

Mixed into models (AssistantConversation) whose records are processed by
background workers. The processing_by_id / processing_since columns
provide UI-visible lock state (who is processing, since when), paired
with with_advisory_lock in the worker for true mutual exclusion.

See Also:

Instance Method Summary collapse

Instance Method Details

#acquire_processing_lock!(party, job_id: nil) ⇒ Boolean

Column-based lock provides UI-visible state (who is processing).
Paired with with_advisory_lock in the worker for true mutual exclusion.

Parameters:

  • party (Party) —

    the party (user/account) acquiring the lock

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

    optional Sidekiq job ID to record in Redis

Returns:

  • (Boolean) —

    true if the lock was acquired, false if another
    non-stale lock is already held



21
22
23
24
25
26
27
28
29
30
31
32
33
 
# File 'app/concerns/models/assistant_conversation_processing_lockable.rb', line 21

def acquire_processing_lock!(party, job_id: nil)
  now = Time.current
  rows = self.class.where(id: id)
             .where('processing_by_id IS NULL OR processing_since < ?', AssistantConversation::LOCK_STALE_AFTER.ago)
             .update_all(processing_by_id: party.id, processing_since: now)
  acquired = rows == 1
  if acquired
    self.processing_by_id = party.id
    self.processing_since = now
    store_processing_job_id!(job_id) if job_id
  end
  acquired
end

#force_processing_lock!(party, job_id: nil) ⇒ Boolean

Unconditionally take the processing lock, bypassing the stale check.
Only safe to call when the caller already holds the advisory lock,
which guarantees no other worker is actively processing.

Parameters:

  • party (Party) —

    the party (user/account) acquiring the lock

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

    optional Sidekiq job ID to record in Redis

Returns:

  • (Boolean) —

    result of update_columns



42
43
44
45
46
47
48
 
# File 'app/concerns/models/assistant_conversation_processing_lockable.rb', line 42

def force_processing_lock!(party, job_id: nil)
  now = Time.current
  update_columns(processing_by_id: party.id, processing_since: now)
  self.processing_by_id = party.id
  self.processing_since = now
  store_processing_job_id!(job_id) if job_id
end

#heartbeat_stale? ⇒ Boolean

Heartbeat has stopped (worker likely killed) but lock hasn't expired yet.
Used by the processing_status endpoint so the client can show an early warning.

Returns:

  • (Boolean) —

    true if the heartbeat is stale but the lock is still valid



76
77
78
79
80
81
 
# File 'app/concerns/models/assistant_conversation_processing_lockable.rb', line 76

def heartbeat_stale?
  processing_by_id.present? &&
    processing_since.present? &&
    processing_since < AssistantConversation::HEARTBEAT_STALE_AFTER.ago &&
    processing_since > AssistantConversation::LOCK_STALE_AFTER.ago
end

#processing? ⇒ Boolean

Whether a non-stale processing lock is currently held.

Returns:

  • (Boolean) —

    true if locked and the lock has not expired



67
68
69
70
 
# File 'app/concerns/models/assistant_conversation_processing_lockable.rb', line 67

def processing?
  processing_by_id.present? && processing_since.present? &&
    processing_since > AssistantConversation::LOCK_STALE_AFTER.ago
end

#processing_job_id ⇒ String?

Sidekiq job ID of the worker currently processing this conversation.
Stored in Redis to avoid JSONB contention with concurrent metadata writes.

Returns:

  • (String, nil) —

    the Sidekiq job ID, or nil if none is recorded



87
88
89
 
# File 'app/concerns/models/assistant_conversation_processing_lockable.rb', line 87

def processing_job_id
  Sidekiq.redis { |conn| conn.get("assistant_processing_jid:#{id}") }
end

#processing_lock_key ⇒ String

Advisory lock key for this conversation.

Returns:

  • (String) —

    key in the form assistant_conversation_<id>



94
95
96
 
# File 'app/concerns/models/assistant_conversation_processing_lockable.rb', line 94

def processing_lock_key
  "assistant_conversation_#{id}"
end

#refresh_processing_ttl! ⇒ Integer

Note:

Uses an atomic DB check to avoid ghost-lock race: a page-load may
have cleared processing_by_id while the worker still holds the
in-memory copy.

Heartbeats the lock: extends the Redis job ID TTL and bumps
processing_since atomically.

Returns:

  • (Integer) —

    number of rows updated (0 if the lock was cleared)



105
106
107
108
109
110
111
 
# File 'app/concerns/models/assistant_conversation_processing_lockable.rb', line 105

def refresh_processing_ttl!
  Sidekiq.redis { |conn| conn.expire("assistant_processing_jid:#{id}", AssistantConversation::PROCESSING_JID_TTL) }
  # Use an atomic DB check to avoid ghost-lock race: a page-load may have
  # cleared processing_by_id while the worker still holds the in-memory copy.
  self.class.where(id: id).where.not(processing_by_id: nil)
      .update_all(processing_since: Time.current)
end

#release_processing_lock! ⇒ Integer

Releases the processing lock, clearing the lock columns and the
Redis-stored Sidekiq job ID.

Returns:

  • (Integer) —

    number of rows updated (1 when the record exists)



54
55
56
57
58
59
60
61
62
 
# File 'app/concerns/models/assistant_conversation_processing_lockable.rb', line 54

def release_processing_lock!
  self.class.where(id: id).update_all(
    processing_by_id: nil,
    processing_since: nil
  )
  clear_processing_job_id!
  self.processing_by_id = nil
  self.processing_since = nil
end