Class: Woods::Coordination::PipelineLock

Inherits:
Object
  • Object
show all
Defined in:
lib/woods/coordination/pipeline_lock.rb

Overview

File-based lock for preventing concurrent pipeline operations.

Creates a lock file with PID and timestamp. Supports stale lock detection for crashed processes.

Examples:

lock = PipelineLock.new(lock_dir: '/tmp', name: 'extraction')
lock.with_lock do
  # extraction runs here
end

Constant Summary collapse

DEFAULT_STALE_TIMEOUT =

1 hour

3600

Instance Method Summary collapse

Constructor Details

#initialize(lock_dir:, name:, stale_timeout: DEFAULT_STALE_TIMEOUT) ⇒ PipelineLock

Returns a new instance of PipelineLock.

Parameters:

  • lock_dir (String) —

    Directory for lock files

  • name (String) —

    Lock name (used as filename prefix)

  • stale_timeout (Integer) (defaults to: DEFAULT_STALE_TIMEOUT) —

    Seconds after which a lock is considered stale



28
29
30
31
32
33
34
 
# File 'lib/woods/coordination/pipeline_lock.rb', line 28

def initialize(lock_dir:, name:, stale_timeout: DEFAULT_STALE_TIMEOUT)
  @lock_dir = lock_dir
  @name = name
  @stale_timeout = stale_timeout
  @lock_path = File.join(lock_dir, "#{name}.lock")
  @held = false
end

Instance Method Details

#acquire ⇒ Boolean

Attempt to acquire the lock.

Returns:

  • (Boolean) —

    true if lock acquired, false if already held



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/woods/coordination/pipeline_lock.rb', line 39

def acquire
  FileUtils.mkdir_p(@lock_dir)

  if File.exist?(@lock_path)
    return false unless stale?
    # Retire the stale lock atomically. A bare rm_f + create here is
    # a TOCTOU race: two processes passing the stale check together
    # could each delete-and-create, the second deleting the first's
    # FRESH lock — both would then "hold" it.
    return false unless retire_stale_lock
  end

  # Atomic lock creation: File::EXCL ensures this fails if file already exists
  File.open(@lock_path, File::WRONLY | File::CREAT | File::EXCL) do |f|
    f.write(lock_content)
  end
  @held = true
  true
rescue Errno::EEXIST
  false
end

#locked? ⇒ Boolean

Whether the lock is currently held by this instance.

Returns:

  • (Boolean) 


114
115
116
 
# File 'lib/woods/coordination/pipeline_lock.rb', line 114

def locked?
  @held && File.exist?(@lock_path)
end

#release ⇒ void

This method returns an undefined value.

Release the lock.

Deletes the lock file only if it still carries this instance's token — a run that outlived stale_timeout may have been legitimately taken over, and deleting unconditionally would drop the new holder's lock.



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/woods/coordination/pipeline_lock.rb', line 69

def release
  return unless @held

  # Clear @held up front so no later failure can leave this instance
  # believing it still holds the lock.
  @held = false

  # Rename first, then inspect: a plain read-then-unlink is a TOCTOU —
  # after we read our own token a takeover could replace the file, and
  # our unlink would then delete the NEW holder's lock. Renaming
  # atomically captures whatever is at the path.
  graveyard = "#{@lock_path}.release.#{Process.pid}.#{SecureRandom.hex(4)}"
  begin
    File.rename(@lock_path, graveyard)
  rescue Errno::ENOENT
    return # already gone
  end

  if own_lock?(graveyard)
    FileUtils.rm_f(graveyard)
  else
    # We were legitimately taken over — put the successor's lock back
    # without clobbering a still-newer holder (see {#restore_lock}).
    restore_lock(graveyard)
  end
end

#with_lock { ... } ⇒ Object

Execute a block while holding the lock.

Yields:

  • Block to execute

Returns:

  • (Object) —

    Return value of the block

Raises:



101
102
103
104
105
106
107
108
109
 
# File 'lib/woods/coordination/pipeline_lock.rb', line 101

def with_lock(&block)
  raise LockError, "Cannot acquire lock '#{@name}' — another process is running" unless acquire

  begin
    block.call
  ensure
    release
  end
end