Class: Dionysus::Producer::KarafkaResponderGenerator

Inherits:
Object
  • Object
show all
Defined in:
lib/dionysus/producer/karafka_responder_generator.rb

Defined Under Namespace

Classes: NullRegistration

Constant Summary collapse

TOMBSTONE = 
nil

Instance Method Summary collapse

Instance Method Details

#generate(config, topic) ⇒ Object 



9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
65
66
67
68
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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/dionysus/producer/karafka_responder_generator.rb', line 9

def generate(config, topic)
  topic_name = topic.to_s
  genesis_topic_name = topic.genesis_to_s if topic.genesis_replica?

  responder_klass = Class.new(Dionysus::Producer::BaseResponder) do
    topic topic_name
    topic genesis_topic_name if topic.genesis_replica?

    define_method :respond do |batch, options = {}|
      config.instrumenter.instrument("dionysus.respond.#{self.class.name}") do
        final_options = {}
        if (partition_key = options.fetch(:partition_key, nil))
          final_options[:partition_key] = partition_key
        end
        if (key = options.fetch(:key, nil))
          final_options[:key] = key
        end

        if genesis_only?(options) && genesis_topic_name.nil?
          raise "cannot execute genesis-only as there is no genesis topic for responder #{self.class.name}"
        end

        if batch.nil?
          unless genesis_only?(options)
            respond_to topic_name, TOMBSTONE, **final_options
            config.event_bus.publish("dionysus.respond", topic_name: topic_name, message: TOMBSTONE,
              options: final_options)
          end
          if topic.genesis_replica?
            respond_to genesis_topic_name, TOMBSTONE, **final_options
            config.event_bus.publish("dionysus.respond", topic_name: genesis_topic_name, message: TOMBSTONE,
              options: final_options)
          end
        else
          message = Array.wrap(batch).map do |event, record_or_records, batch_options|
            records = Array.wrap(record_or_records)
            return if records.empty?

            record = records.sample

            # the offset alone is publish order, and a message published later can carry an
            # earlier snapshot, so consumers need to know when this payload was actually read
            serialized_at = config.include_serialized_at_in_payload ? Time.now.utc : nil
            payload = serialize_consistently(records, topic, batch_options)

            event_payload = {
              event: event,
              model_name: record.model_name.name,
              data: payload
            }
            event_payload[:serialized_at] = serialized_at.iso8601(6) if serialized_at
            event_payload
          end
          unless genesis_only?(options)
            respond_to topic_name, { message: message }, **final_options
            config.event_bus.publish("dionysus.respond", topic_name: topic_name, message: message,
              options: final_options)
          end
          if topic.genesis_replica?
            respond_to genesis_topic_name, { message: message }, **final_options
            config.event_bus.publish("dionysus.respond", topic_name: genesis_topic_name, message: message,
              options: final_options)
          end
        end
      end
    end

    private

    # A serializer reads a record's own columns and then queries its associations, so anything
    # committed in between lands in the payload beside a timestamp taken before it - the payload
    # describes no single moment. Consumers rank on that timestamp and then compare it against
    # what they have stored, so a payload whose timestamp predates its own contents loses the
    # comparison and is discarded, taking the has_many records embedded in it along with it.
    #
    # Measured on the affected topic in production: serialization spans 95ms at the median and up
    # to 3.5s, and 4.7% of records receiving more than one message had the surviving message
    # report a timestamp older than one already published.
    #
    # So serialize, then check whether the records moved underneath it. If they did, the payload
    # is not a snapshot of anything: reload and serialize again. Retries are bounded, and the last
    # attempt is published rather than dropped - a payload that may be torn still beats no message.
    define_method :serialize_consistently do |records, current_topic, batch_options|
      next serialize_to_payload(records, current_topic, batch_options) unless config.publish_consistent_snapshots

      max_attempts = config.max_snapshot_attempts
      attempts = 0
      loop do
        before = snapshot_of(records)
        payload = serialize_to_payload(records, current_topic, batch_options)
        attempts += 1

        # nothing to compare against, so the guard did not run on this message at all
        break instrument_snapshot("unsupported", attempts, records, current_topic, payload) if before.nil?
        if before == committed_snapshot_of(records)
          break instrument_snapshot("consistent", attempts, records, current_topic, payload)
        end
        if attempts >= max_attempts
          # published anyway: a payload that may be torn beats no message. Nothing downstream can
          # tell this apart from a clean one - the payload carries no marker and the consumer sees
          # an ordinary message - so this counter is the only place the outcome is ever visible.
          break instrument_snapshot("exhausted", attempts, records, current_topic, payload)
        end

        records.each { |record| record.reload if record.is_a?(ActiveRecord::Base) && record.persisted? }
      end
    end

    # One counter rather than several: the denominator, the retry distribution and the failure
    # rate all have to come from the same series or none of them can be read as a rate.
    define_method :instrument_snapshot do |result, attempts, records, current_topic, payload|
      config.instrumenter.increment("dionysus.publish.consistent_snapshot",
        tags: ["result:#{result}", "attempts:#{attempts}", "topic:#{current_topic}",
          "model:#{records.first.class}"])
      payload
    end

    # nil means there is nothing to compare - a record without timestamps, or not a record at all
    define_method :snapshot_of do |records|
      stamps = records.map { |record| record.updated_at if record.respond_to?(:updated_at) }
      stamps.any?(&:nil?) ? nil : stamps
    end

    define_method :committed_snapshot_of do |records|
      records.map do |record|
        next record.updated_at unless record.is_a?(ActiveRecord::Base) && record.persisted?

        # publishing runs inside the request when publish_after_commit is on, where the query
        # cache is live on this connection - a cached read here would report that nothing moved
        # and hand back the torn payload the check exists to catch
        record.class.uncached do
          record.class.where(record.class.primary_key => record.id).pick(:updated_at)
        end
      end
    end

    define_method :serialize_to_payload do |records, current_topic, batch_options|
      if batch_options.to_h[:serialize] == false
        records.map(&:as_json)
      else
        record = records.sample

        model_klass = record.class
        dependencies = current_topic
          .models
          .find(-> { NullRegistration.new }) { |model_registration| model_registration.model_klass == model_klass }
          .options
          .to_h
          .fetch(:with, [])

        current_topic.serializer_klass.serialize(records, dependencies: dependencies)
      end
    end

    define_method :genesis_only? do |options|
      options.fetch(:genesis_only, false) == true
    end
  end

  responder_klass.instance_exec(topic) do |dionysus_topic|
    define_singleton_method :publisher_of? do |model_klass|
      dionysus_topic.publishes_model?(model_klass)
    end

    define_singleton_method :publisher_for_topic? do |current_topic|
      if dionysus_topic.genesis_replica?
        dionysus_topic.to_s == current_topic.to_s || dionysus_topic.genesis_to_s == current_topic.to_s
      else
        dionysus_topic.to_s == current_topic.to_s
      end
    end

    define_singleton_method :publisher_of_model_for_topic? do |model_klass, current_topic|
      dionysus_topic.publishes_model?(model_klass) && publisher_for_topic?(current_topic)
    end

    define_singleton_method :partition_key do
      dionysus_topic.partition_key
    end

    define_singleton_method :primary_topic do
      responder_klass.topics.values.first
    end
  end

  responder_klass_name = "#{topic.to_s.classify}Responder"

  Dionysus.send(:remove_const, responder_klass_name) if Dionysus.const_defined?(responder_klass_name)
  Dionysus.const_set(responder_klass_name, responder_klass)
  responder_klass
end