Class: Mammoth::DeliveryProgressCoordinator

Inherits:
Object
  • Object
show all
Defined in:
lib/mammoth/delivery_progress_coordinator.rb

Overview

Coordinates durable delivery progress across out-of-order workers.

Work is registered in source order before dispatch. Completion may arrive in any order, but a checkpoint and upstream acknowledgement advance only when every item in the oldest closed source group has a durable outcome. A source group is normally one PostgreSQL transaction.

Defined Under Namespace

Classes: Entry, Group

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(checkpoint_store:, source_name:, slot_name:, publication_name:, acknowledger: nil, position_resolver: nil) ⇒ DeliveryProgressCoordinator

Returns a new instance of DeliveryProgressCoordinator.

Parameters:

  • checkpoint_store (Mammoth::CheckpointStore) —

    durable checkpoint persistence

  • source_name (String) —

    logical source name

  • slot_name (String) —

    PostgreSQL replication slot name

  • publication_name (String) —

    publication name

  • acknowledger (#call, nil) (defaults to: nil) —

    upstream durable-progress acknowledgement

  • position_resolver (#call, nil) (defaults to: nil) —

    source-owned durable position resolver



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/mammoth/delivery_progress_coordinator.rb', line 47

def initialize(checkpoint_store:, source_name:, slot_name:, publication_name:, acknowledger: nil,
               position_resolver: nil)
  @checkpoint_store = checkpoint_store
  @source_name = source_name
  @slot_name = slot_name
  @publication_name = publication_name
  @acknowledger = acknowledger
  @position_resolver = position_resolver
  # rubocop:disable Layout/LeadingCommentSpace -- Steep inline type syntax requires `#:`.
  @groups = [] #: Array[Group]
  @entries_by_work = {} #: Hash[untyped, Array[Entry]]
  # rubocop:enable Layout/LeadingCommentSpace
  @entries_by_work.compare_by_identity
  @mutex = Mutex.new
end

Instance Attribute Details

#checkpoint_store ⇒ Object (readonly)

Returns the value of attribute checkpoint_store.



39
40
41
 
# File 'lib/mammoth/delivery_progress_coordinator.rb', line 39

def checkpoint_store
  @checkpoint_store
end

#publication_name ⇒ Object (readonly)

Returns the value of attribute publication_name.



39
40
41
 
# File 'lib/mammoth/delivery_progress_coordinator.rb', line 39

def publication_name
  @publication_name
end

#slot_name ⇒ Object (readonly)

Returns the value of attribute slot_name.



39
40
41
 
# File 'lib/mammoth/delivery_progress_coordinator.rb', line 39

def slot_name
  @slot_name
end

#source_name ⇒ Object (readonly)

Returns the value of attribute source_name.



39
40
41
 
# File 'lib/mammoth/delivery_progress_coordinator.rb', line 39

def source_name
  @source_name
end

Instance Method Details

#complete(work) ⇒ String?

Mark a work item durably resolved and advance all newly contiguous groups.

Parameters:

  • work (CDC::Core::ChangeEvent, CDC::Core::TransactionEnvelope) —

    work item

Returns:

  • (String, nil) —

    latest source position advanced by this call



83
84
85
86
87
88
89
90
91
 
# File 'lib/mammoth/delivery_progress_coordinator.rb', line 83

def complete(work)
  @mutex.synchronize do
    entry = pending_entry_for(work)
    raise ReplicationError, "delivery progress completed before registration" unless entry

    entry.completed = true
    advance_contiguous_groups
  end
end

#finalize ⇒ String?

Close and advance a final source group after a clean end-of-stream.

Returns:

  • (String, nil) —

    latest source position advanced by this call



96
97
98
99
100
101
 
# File 'lib/mammoth/delivery_progress_coordinator.rb', line 96

def finalize
  @mutex.synchronize do
    @groups.last.closed = true if @groups.last
    advance_contiguous_groups
  end
end

#register(work, group_end:) ⇒ void

This method returns an undefined value.

Register work before dispatch.

Parameters:

  • work (CDC::Core::ChangeEvent, CDC::Core::TransactionEnvelope) —

    work item

  • group_end (Boolean) —

    whether this is the final item in its source group



68
69
70
71
72
73
74
75
76
77
 
# File 'lib/mammoth/delivery_progress_coordinator.rb', line 68

def register(work, group_end:)
  @mutex.synchronize do
    group = current_group
    entry = Entry.new(work: work, source_position: source_position(work), completed: false)
    group.items << entry
    (@entries_by_work[work] ||= []) << entry # steep:ignore
    group.closed = true if group_end
  end
  nil
end