Class: Shikibu::Storage::SequelStorage

Inherits:
Object
  • Object
show all
Defined in:
lib/shikibu/storage/sequel_storage.rb

Overview

Sequel-based storage implementation

Constant Summary collapse

DEFAULT_LOCK_TIMEOUT =

5 minutes

300

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(database_url, auto_migrate: false) ⇒ SequelStorage

Returns a new instance of SequelStorage.



16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 16

def initialize(database_url, auto_migrate: false)
  @database_url = database_url
  @db = Sequel.connect(database_url)
  @notify_enabled = false

  # Enable extensions based on database type
  configure_database

  # Apply migrations if requested
  Migrations.apply(@db) if auto_migrate
end

Instance Attribute Details

#db ⇒ Object (readonly)

Returns the value of attribute db.



11
12
13
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 11

def db
  @db
end

#notify_enabled ⇒ Object

Returns the value of attribute notify_enabled.



12
13
14
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 12

def notify_enabled
  @notify_enabled
end

Instance Method Details

#add_outbox_event(event_id:, event_type:, event_source:, event_data:, data_type: DataType::JSON) ⇒ Object

============================================

Outbox



610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 610

def add_outbox_event(event_id:, event_type:, event_source:, event_data:, data_type: DataType::JSON)
  if data_type == DataType::BINARY
    @db[:outbox_events].insert(
      event_id: event_id,
      event_type: event_type,
      event_source: event_source,
      data_type: data_type,
      event_data_binary: Sequel.blob(event_data),
      status: OutboxStatus::PENDING,
      created_at: current_timestamp
    )
  else
    @db[:outbox_events].insert(
      event_id: event_id,
      event_type: event_type,
      event_source: event_source,
      data_type: data_type,
      event_data: serialize_json(event_data),
      status: OutboxStatus::PENDING,
      created_at: current_timestamp
    )
  end

  # Send NOTIFY for new outbox event (deferred if in transaction)
  notify_payload = { evt_id: event_id, evt_type: event_type }
  if in_transaction?
    register_post_commit_callback do
      send_notify(Notify::Channel::OUTBOX_PENDING, notify_payload)
    end
  else
    send_notify(Notify::Channel::OUTBOX_PENDING, notify_payload)
  end
end

#append_history(instance_id:, activity_id:, event_type:, event_data:, data_type: DataType::JSON) ⇒ Object

============================================

Workflow History



286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 286

def append_history(instance_id:, activity_id:, event_type:, event_data:, data_type: DataType::JSON)
  if data_type == DataType::BINARY
    @db[:workflow_history].insert(
      instance_id: instance_id,
      activity_id: activity_id,
      event_type: event_type,
      data_type: data_type,
      event_data_binary: Sequel.blob(event_data),
      created_at: current_timestamp
    )
  else
    @db[:workflow_history].insert(
      instance_id: instance_id,
      activity_id: activity_id,
      event_type: event_type,
      data_type: data_type,
      event_data: serialize_json(event_data),
      created_at: current_timestamp
    )
  end
rescue Sequel::UniqueConstraintViolation
  # Activity already recorded (idempotent)
  nil
end

#archive_history(instance_id) ⇒ Object 



318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 318

def archive_history(instance_id)
  now = current_timestamp

  # Copy to archive
  @db[:workflow_history_archive].insert(
    @db[:workflow_history]
      .where(instance_id: instance_id)
      .select(
        :instance_id, :activity_id, :event_type, :data_type,
        :event_data, :event_data_binary, :created_at,
        Sequel.lit("'#{now}'").as(:archived_at)
      )
  )

  # Delete from main history
  @db[:workflow_history].where(instance_id: instance_id).delete
end

#claim_message(message_id:, instance_id:) ⇒ Object 



525
526
527
528
529
530
531
532
533
534
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 525

def claim_message(message_id:, instance_id:)
  @db[:channel_message_claims].insert(
    message_id: message_id,
    instance_id: instance_id,
    claimed_at: current_timestamp
  )
  true
rescue Sequel::UniqueConstraintViolation
  false
end

#cleanup_old_channel_messages(retention_days:) ⇒ Integer

Cleanup old channel messages older than retention_days

Parameters:

  • Number of days to retain messages

Returns:

  • Number of deleted messages



782
783
784
785
786
787
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 782

def cleanup_old_channel_messages(retention_days:)
  cutoff_time = Time.now - (retention_days * 24 * 60 * 60)
  @db[:channel_messages]
    .where { published_at < cutoff_time }
    .delete
end

#clear_compensations(instance_id) ⇒ Object 



366
367
368
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 366

def clear_compensations(instance_id)
  @db[:workflow_compensations].where(instance_id: instance_id).delete
end

#close ⇒ Object 



28
29
30
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 28

def close
  @db.disconnect
end

#create_instance(instance_id:, workflow_name:, source_hash:, owner_service:, input_data:, status: Status::RUNNING) ⇒ Object

============================================

Workflow Instances



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 116

def create_instance(
  instance_id:,
  workflow_name:,
  source_hash:,
  owner_service:,
  input_data:,
  status: Status::RUNNING
)
  @db[:workflow_instances].insert(
    instance_id: instance_id,
    workflow_name: workflow_name,
    source_hash: source_hash,
    owner_service: owner_service,
    framework: FRAMEWORK,
    status: status,
    input_data: serialize_json(input_data),
    started_at: current_timestamp,
    updated_at: current_timestamp
  )
  instance_id
end

#deliver_channel_message(instance_id:, channel:, message_id:, data:, metadata:, worker_id:) ⇒ Hash?

Deliver a channel message directly to a specific workflow instance Used for Point-to-Point delivery (e.g., shikibuinstanceid targeting)

Parameters:

  • Target workflow instance ID

  • Channel name (event type)

  • Message ID

  • Message data

  • Message metadata

  • Worker ID for locking

Returns:

  • Delivery result or nil if delivery failed



554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 554

def deliver_channel_message(instance_id:, channel:, message_id:, data:, metadata:, worker_id:)
  # Try to acquire lock on the workflow instance
  return nil unless try_acquire_lock(instance_id, worker_id)

  begin
    result = @db.transaction do
      # Find waiting subscription for this channel
      sub = @db[:channel_subscriptions]
            .where(instance_id: instance_id, channel: channel)
            .where { Sequel.~(activity_id: nil) }
            .first

      # No waiting subscription found
      next nil unless sub

      activity_id = sub[:activity_id]

      # Record message received in history (for deterministic replay)
      append_history(
        instance_id: instance_id,
        activity_id: activity_id,
        event_type: EventType::CHANNEL_MESSAGE_RECEIVED,
        event_data: {
          channel: channel,
          message_id: message_id,
          data: data,
          metadata: metadata,
          published_at: Time.now.iso8601
        }
      )

      # Clear activity_id from subscription (no longer waiting)
      @db[:channel_subscriptions]
        .where(instance_id: instance_id, channel: channel)
        .update(activity_id: nil, timeout_at: nil)

      # Update workflow instance status to running
      update_instance_status(instance_id, Status::RUNNING)

      { instance_id: instance_id, activity_id: activity_id }
    end

    # Send NOTIFY for workflow resumable (outside transaction for PostgreSQL)
    send_notify(Notify::Channel::WORKFLOW_RESUMABLE, { wf_id: instance_id }) if result

    result
  ensure
    # Always release the lock
    release_lock(instance_id, worker_id)
  end
end

#find_expired_timers(limit: 100) ⇒ Object 



401
402
403
404
405
406
407
408
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 401

def find_expired_timers(limit: 100)
  now = Time.now
  @db[:workflow_timer_subscriptions]
    .where { expires_at <= now }
    .order(:expires_at)
    .limit(limit)
    .map(&:to_h)
end

#find_resumable_workflows(limit: 10) ⇒ Object 



167
168
169
170
171
172
173
174
175
176
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 167

def find_resumable_workflows(limit: 10)
  @db[:workflow_instances]
    .where(status: Status::RUNNING)
    .where(locked_by: nil)
    .where(framework: FRAMEWORK)
    .order(:updated_at)
    .limit(limit)
    .select(:instance_id, :workflow_name, :source_hash)
    .map(&:to_h)
end

#find_stale_locked_workflows(stale_threshold_seconds: 300) ⇒ Object 



178
179
180
181
182
183
184
185
186
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 178

def find_stale_locked_workflows(stale_threshold_seconds: 300)
  threshold = Time.now - stale_threshold_seconds
  @db[:workflow_instances]
    .where(status: Status::RUNNING)
    .where { lock_expires_at < threshold }
    .where(framework: FRAMEWORK)
    .select(:instance_id, :workflow_name, :locked_by)
    .map(&:to_h)
end

#find_timed_out_subscriptions(limit: 100) ⇒ Object 



516
517
518
519
520
521
522
523
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 516

def find_timed_out_subscriptions(limit: 100)
  now = Time.now
  @db[:channel_subscriptions]
    .where { timeout_at <= now }
    .where { Sequel.~(timeout_at: nil) }
    .limit(limit)
    .map(&:to_h)
end

#find_waiting_subscriptions(channel:, limit: 100) ⇒ Object 



508
509
510
511
512
513
514
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 508

def find_waiting_subscriptions(channel:, limit: 100)
  @db[:channel_subscriptions]
    .where(channel: channel)
    .where { Sequel.~(activity_id: nil) }
    .limit(limit)
    .map(&:to_h)
end

#get_channel_mode(channel) ⇒ String?

Get the mode for a channel (from first subscription)

Parameters:

  • Channel name

Returns:

  • The mode or nil if no subscriptions exist



500
501
502
503
504
505
506
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 500

def get_channel_mode(channel)
  row = @db[:channel_subscriptions]
        .where(channel: channel)
        .select(:mode)
        .first
  row&.dig(:mode)
end

#get_compensations(instance_id) ⇒ Object 



350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 350

def get_compensations(instance_id)
  @db[:workflow_compensations]
    .where(instance_id: instance_id)
    .order(Sequel.desc(:created_at))
    .map do |row|
      {
        id: row[:id],
        instance_id: row[:instance_id],
        activity_id: row[:activity_id],
        activity_name: row[:activity_name],
        args: parse_json(row[:args]),
        created_at: row[:created_at]
      }
    end
end

#get_history(instance_id) ⇒ Object 



311
312
313
314
315
316
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 311

def get_history(instance_id)
  @db[:workflow_history]
    .where(instance_id: instance_id)
    .order(:id)
    .map { |row| deserialize_history_event(row) }
end

#get_instance(instance_id) ⇒ Object 



138
139
140
141
142
143
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 138

def get_instance(instance_id)
  row = @db[:workflow_instances].where(instance_id: instance_id).first
  return nil unless row

  deserialize_instance(row)
end

#get_next_message(channel:, mode:, instance_id:, cursor_id: nil) ⇒ Object 



536
537
538
539
540
541
542
543
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 536

def get_next_message(channel:, mode:, instance_id:, cursor_id: nil)
  case mode
  when ChannelMode::COMPETING
    get_next_competing_message(channel, instance_id)
  when ChannelMode::BROADCAST
    get_next_broadcast_message(channel, instance_id, cursor_id)
  end
end

#get_pending_outbox_events(limit: 100) ⇒ Object 



644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 644

def get_pending_outbox_events(limit: 100)
  if supports_skip_locked?
    # PostgreSQL/MySQL: Use SELECT FOR UPDATE SKIP LOCKED
    # This allows multiple workers to fetch different events without blocking
    @db.transaction do
      rows = @db[:outbox_events]
             .where(status: [OutboxStatus::PENDING, OutboxStatus::FAILED])
             .order(:created_at)
             .limit(limit)
             .for_update
             .skip_locked
             .all

      return [] if rows.empty?

      # Mark as processing to prevent duplicate fetches
      event_ids = rows.map { |r| r[:event_id] }
      @db[:outbox_events]
        .where(event_id: event_ids)
        .update(status: OutboxStatus::PROCESSING)

      rows.map { |row| deserialize_outbox_event(row) }
    end
  else
    # SQLite: Simple query (table-level locking)
    @db[:outbox_events]
      .where(status: [OutboxStatus::PENDING, OutboxStatus::FAILED])
      .order(:created_at)
      .limit(limit)
      .map { |row| deserialize_outbox_event(row) }
  end
end

#get_subscription(instance_id:, channel:) ⇒ Object 



491
492
493
494
495
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 491

def get_subscription(instance_id:, channel:)
  @db[:channel_subscriptions]
    .where(instance_id: instance_id, channel: channel)
    .first
end

#get_workflow_definition(workflow_name:, source_hash:) ⇒ Object 



106
107
108
109
110
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 106

def get_workflow_definition(workflow_name:, source_hash:)
  @db[:workflow_definitions]
    .where(workflow_name: workflow_name, source_hash: source_hash)
    .first
end

#in_transaction? ⇒ Boolean

Returns:



75
76
77
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 75

def in_transaction?
  @db.in_transaction?
end

#list_instances(limit: 100, offset: 0, status_filter: nil, workflow_name: nil) ⇒ Object 



158
159
160
161
162
163
164
165
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 158

def list_instances(limit: 100, offset: 0, status_filter: nil, workflow_name: nil)
  query = @db[:workflow_instances].order(Sequel.desc(:updated_at))
  query = query.where(status: status_filter) if status_filter
  query = query.where(workflow_name: workflow_name) if workflow_name
  query = query.limit(limit).offset(offset)

  query.map { |row| deserialize_instance(row) }
end

#lock_held_by?(instance_id, worker_id) ⇒ Boolean

Returns:



269
270
271
272
273
274
275
276
277
278
279
280
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 269

def lock_held_by?(instance_id, worker_id)
  row = @db[:workflow_instances]
        .where(instance_id: instance_id)
        .select(:locked_by, :lock_expires_at)
        .first

  return false unless row
  return false if row[:locked_by] != worker_id
  return false if row[:lock_expires_at] && row[:lock_expires_at] < Time.now

  true
end

#mark_outbox_expired(event_id, error_message) ⇒ Object 



696
697
698
699
700
701
702
703
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 696

def mark_outbox_expired(event_id, error_message)
  @db[:outbox_events]
    .where(event_id: event_id)
    .update(
      status: OutboxStatus::EXPIRED,
      last_error: error_message
    )
end

#mark_outbox_failed(event_id, error_message) ⇒ Object 



686
687
688
689
690
691
692
693
694
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 686

def mark_outbox_failed(event_id, error_message)
  @db[:outbox_events]
    .where(event_id: event_id)
    .update(
      status: OutboxStatus::FAILED,
      last_error: error_message,
      retry_count: Sequel[:retry_count] + 1
    )
end

#mark_outbox_invalid(event_id, error_message) ⇒ Object 



705
706
707
708
709
710
711
712
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 705

def mark_outbox_invalid(event_id, error_message)
  @db[:outbox_events]
    .where(event_id: event_id)
    .update(
      status: OutboxStatus::INVALID,
      last_error: error_message
    )
end

#mark_outbox_published(event_id) ⇒ Object 



677
678
679
680
681
682
683
684
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 677

def mark_outbox_published(event_id)
  @db[:outbox_events]
    .where(event_id: event_id)
    .update(
      status: OutboxStatus::PUBLISHED,
      published_at: current_timestamp
    )
end

#notify_workflow_resumable(instance_id, workflow_name = nil) ⇒ Object

Send workflow resumable notification

Parameters:

  • Workflow instance ID

  • (defaults to: nil)

    Optional workflow name



775
776
777
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 775

def notify_workflow_resumable(instance_id, workflow_name = nil)
  send_notify(Notify::Channel::WORKFLOW_RESUMABLE, { wf_id: instance_id, wf_name: workflow_name })
end

#publish_message(channel:, data:, metadata: nil, data_type: DataType::JSON) ⇒ Object

============================================

Channel Messages



420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 420

def publish_message(channel:, data:, metadata: nil, data_type: DataType::JSON)
  message_id = SecureRandom.uuid

  if data_type == DataType::BINARY
    @db[:channel_messages].insert(
      channel: channel,
      message_id: message_id,
      data_type: data_type,
      data_binary: Sequel.blob(data),
      metadata: metadata ? serialize_json(metadata) : nil,
      published_at: current_timestamp
    )
  else
    @db[:channel_messages].insert(
      channel: channel,
      message_id: message_id,
      data_type: data_type,
      data: serialize_json(data),
      metadata: metadata ? serialize_json(metadata) : nil,
      published_at: current_timestamp
    )
  end

  # Send NOTIFY for new message (deferred if in transaction)
  notify_payload = { ch: channel, msg_id: message_id }
  if in_transaction?
    register_post_commit_callback do
      send_notify(Notify::Channel::CHANNEL_MESSAGE, notify_payload)
    end
  else
    send_notify(Notify::Channel::CHANNEL_MESSAGE, notify_payload)
  end

  message_id
end

#push_compensation(instance_id:, activity_id:, activity_name:, args:) ⇒ Object

============================================

Compensations



340
341
342
343
344
345
346
347
348
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 340

def push_compensation(instance_id:, activity_id:, activity_name:, args:)
  @db[:workflow_compensations].insert(
    instance_id: instance_id,
    activity_id: activity_id,
    activity_name: activity_name,
    args: serialize_json(args),
    created_at: current_timestamp
  )
end

#refresh_lock(instance_id, worker_id, timeout: DEFAULT_LOCK_TIMEOUT) ⇒ Object 



255
256
257
258
259
260
261
262
263
264
265
266
267
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 255

def refresh_lock(instance_id, worker_id, timeout: DEFAULT_LOCK_TIMEOUT)
  now = Time.now
  expires_at = now + timeout

  affected = @db[:workflow_instances]
             .where(instance_id: instance_id, locked_by: worker_id)
             .update(
               locked_at: now,
               lock_expires_at: expires_at
             )

  affected.positive?
end

#refresh_system_lock(lock_name, worker_id, timeout: 30) ⇒ Object 



754
755
756
757
758
759
760
761
762
763
764
765
766
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 754

def refresh_system_lock(lock_name, worker_id, timeout: 30)
  now = Time.now
  expires_at = now + timeout

  affected = @db[:system_locks]
             .where(lock_name: lock_name, locked_by: worker_id)
             .update(
               locked_at: now,
               lock_expires_at: expires_at
             )

  affected.positive?
end

#register_post_commit_callback(&block) ⇒ Object

Register a callback to be executed after successful commit of the outermost transaction

Parameters:

  • The callback to execute after commit

Raises:

  • If called outside of a transaction



45
46
47
48
49
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 45

def register_post_commit_callback(&block)
  raise 'Not in transaction' unless in_transaction?

  transaction_state[:callbacks] << block
end

#register_timer(instance_id:, timer_id:, expires_at:, activity_id: nil) ⇒ Object

============================================

Timer Subscriptions



374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 374

def register_timer(instance_id:, timer_id:, expires_at:, activity_id: nil)
  if mysql?
    # MySQL: Use INSERT ... ON DUPLICATE KEY UPDATE
    @db[:workflow_timer_subscriptions].on_duplicate_key_update(
      expires_at: expires_at,
      activity_id: activity_id
    ).insert(
      instance_id: instance_id,
      timer_id: timer_id,
      expires_at: expires_at,
      activity_id: activity_id,
      created_at: current_timestamp
    )
  else
    @db[:workflow_timer_subscriptions].insert_conflict(
      target: i[instance_id timer_id],
      update: { expires_at: expires_at, activity_id: activity_id }
    ).insert(
      instance_id: instance_id,
      timer_id: timer_id,
      expires_at: expires_at,
      activity_id: activity_id,
      created_at: current_timestamp
    )
  end
end

#release_lock(instance_id, worker_id) ⇒ Object 



244
245
246
247
248
249
250
251
252
253
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 244

def release_lock(instance_id, worker_id)
  @db[:workflow_instances]
    .where(instance_id: instance_id, locked_by: worker_id)
    .update(
      locked_by: nil,
      locked_at: nil,
      lock_expires_at: nil,
      lock_timeout_seconds: nil
    )
end

#release_system_lock(lock_name, worker_id) ⇒ Object 



748
749
750
751
752
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 748

def release_system_lock(lock_name, worker_id)
  @db[:system_locks]
    .where(lock_name: lock_name, locked_by: worker_id)
    .delete
end

#remove_timer(instance_id:, timer_id:) ⇒ Object 



410
411
412
413
414
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 410

def remove_timer(instance_id:, timer_id:)
  @db[:workflow_timer_subscriptions]
    .where(instance_id: instance_id, timer_id: timer_id)
    .delete
end

#save_workflow_definition(workflow_name:, source_hash:, source_code:) ⇒ Object

============================================

Workflow Definitions



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 83

def save_workflow_definition(workflow_name:, source_hash:, source_code:)
  if mysql?
    # MySQL: Use INSERT IGNORE + UPDATE
    @db[:workflow_definitions].insert_ignore.insert(
      workflow_name: workflow_name,
      source_hash: source_hash,
      source_code: source_code,
      created_at: current_timestamp
    )
  else
    # PostgreSQL/SQLite: Use ON CONFLICT
    @db[:workflow_definitions].insert_conflict(
      target: i[workflow_name source_hash],
      update: { source_code: source_code }
    ).insert(
      workflow_name: workflow_name,
      source_hash: source_hash,
      source_code: source_code,
      created_at: current_timestamp
    )
  end
end

#subscribe_to_channel(instance_id:, channel:, mode:, activity_id: nil, timeout_at: nil) ⇒ Object 



456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 456

def subscribe_to_channel(instance_id:, channel:, mode:, activity_id: nil, timeout_at: nil)
  if mysql?
    @db[:channel_subscriptions].on_duplicate_key_update(
      mode: mode,
      activity_id: activity_id,
      timeout_at: timeout_at
    ).insert(
      instance_id: instance_id,
      channel: channel,
      mode: mode,
      activity_id: activity_id,
      timeout_at: timeout_at,
      subscribed_at: current_timestamp
    )
  else
    @db[:channel_subscriptions].insert_conflict(
      target: i[instance_id channel],
      update: { mode: mode, activity_id: activity_id, timeout_at: timeout_at }
    ).insert(
      instance_id: instance_id,
      channel: channel,
      mode: mode,
      activity_id: activity_id,
      timeout_at: timeout_at,
      subscribed_at: current_timestamp
    )
  end
end

#transaction ⇒ Object 



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 51

def transaction(&)
  state = transaction_state
  state[:depth] += 1
  committed = false

  begin
    result = @db.transaction(&)
    committed = true

    # Execute callbacks only after successful commit of outermost transaction
    if state[:depth] == 1
      callbacks = state[:callbacks].dup
      state[:callbacks].clear
      callbacks.each(&:call)
    end

    result
  ensure
    state[:depth] -= 1
    # Clear callbacks on rollback (when not committed) at outermost level
    state[:callbacks].clear if state[:depth].zero? && !committed
  end
end

#transaction_state ⇒ Object

Thread-local storage for transaction state Tracks nesting depth and post-commit callbacks



38
39
40
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 38

def transaction_state
  Thread.current[:shikibu_tx_state] ||= { depth: 0, callbacks: [] }
end

#try_acquire_lock(instance_id, worker_id, timeout: DEFAULT_LOCK_TIMEOUT) ⇒ Object

============================================

Distributed Locking



192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 192

def try_acquire_lock(instance_id, worker_id, timeout: DEFAULT_LOCK_TIMEOUT)
  now = Time.now
  expires_at = now + timeout

  if supports_skip_locked?
    # PostgreSQL/MySQL: Use SELECT FOR UPDATE SKIP LOCKED
    # This prevents blocking when another worker holds the lock
    @db.transaction do
      row = @db[:workflow_instances]
            .where(instance_id: instance_id)
            .for_update
            .skip_locked
            .first

      # Row is locked by another transaction, skip
      return false unless row

      # Check if lock is available (not locked or expired)
      return false if row[:locked_by] && row[:lock_expires_at] && row[:lock_expires_at] > now

      @db[:workflow_instances]
        .where(instance_id: instance_id)
        .update(
          locked_by: worker_id,
          locked_at: now,
          lock_expires_at: expires_at,
          lock_timeout_seconds: timeout
        )

      true
    end
  else
    # SQLite: Use atomic UPDATE (table-level locking)
    affected = @db[:workflow_instances]
               .where(instance_id: instance_id)
               .where(
                 Sequel.|(
                   { locked_by: nil },
                   Sequel.lit('lock_expires_at < ?', now)
                 )
               )
               .update(
                 locked_by: worker_id,
                 locked_at: now,
                 lock_expires_at: expires_at,
                 lock_timeout_seconds: timeout
               )

    affected.positive?
  end
end

#try_acquire_system_lock(lock_name, worker_id, timeout: 30) ⇒ Object

============================================

System Locks



718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 718

def try_acquire_system_lock(lock_name, worker_id, timeout: 30)
  now = Time.now
  expires_at = now + timeout

  # Try to insert new lock
  begin
    @db[:system_locks].insert(
      lock_name: lock_name,
      locked_by: worker_id,
      locked_at: now,
      lock_expires_at: expires_at
    )
    return true
  rescue Sequel::UniqueConstraintViolation
    # Lock exists, try to acquire if expired
  end

  # Try to take over expired lock
  affected = @db[:system_locks]
             .where(lock_name: lock_name)
             .where { lock_expires_at < now }
             .update(
               locked_by: worker_id,
               locked_at: now,
               lock_expires_at: expires_at
             )

  affected.positive?
end

#unsubscribe_from_channel(instance_id:, channel:) ⇒ Object 



485
486
487
488
489
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 485

def unsubscribe_from_channel(instance_id:, channel:)
  @db[:channel_subscriptions]
    .where(instance_id: instance_id, channel: channel)
    .delete
end

#update_instance_status(instance_id, status, output_data: nil, current_activity_id: nil) ⇒ Object 



145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/shikibu/storage/sequel_storage.rb', line 145

def update_instance_status(instance_id, status, output_data: nil, current_activity_id: nil)
  updates = {
    status: status,
    updated_at: current_timestamp
  }
  updates[:output_data] = serialize_json(output_data) if output_data
  updates[:current_activity_id] = current_activity_id if current_activity_id

  @db[:workflow_instances]
    .where(instance_id: instance_id)
    .update(updates)
end