Class: DiscordStore::Journal

Inherits:
Object
  • Object
show all
Defined in:
lib/discord_store/journal.rb

Overview

The write-ahead log behind the ActiveRecord adapter.

Each entry is a SQL statement plus its bind values, which is logical replication in the plainest possible form: replaying the statements in order against an empty database reproduces the database. Postgres and MySQL both ship a version of this idea; the only novelty here is that the log segment is a Discord channel.

Statements are journalled with their binds already materialised, because a prepared statement handle means nothing to a reader on another machine three months from now.

Constant Summary collapse

STREAM = 
"wal"
CURSOR_TABLE = 
"discord_store_journal"
MODES = 
%i[sync async off].freeze
BINARY_TAG =

Tags for values JSON cannot carry losslessly.

"__b"
TIME_TAG = 
"__t"
DATE_TAG = 
"__D"
DECIMAL_TAG = 
"__d"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config:, mode: :sync, flush_interval: 1.0, max_buffer: 200, http: nil) ⇒ Journal

Returns a new instance of Journal.

Parameters:

  • config (Configuration)
  • mode (Symbol) (defaults to: :sync) —

    :sync, :async or :off

  • flush_interval (Numeric) (defaults to: 1.0) —

    seconds, for :async

  • max_buffer (Integer) (defaults to: 200) —

    entries, for :async

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

    injectable transport

Raises:



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/discord_store/journal.rb', line 38

def initialize(config:, mode: :sync, flush_interval: 1.0, max_buffer: 200, http: nil)
  raise ConfigurationError, "journal_mode must be one of #{MODES.join(", ")}" unless MODES.include?(mode)

  @mode = mode
  @flush_interval = flush_interval
  @max_buffer = max_buffer
  @buffer = []
  @mutex = Mutex.new
  @local_cursor = nil

  return if mode == :off

  @client = Client.new(config: config, http: http)
  @log = @client.log
  start_flusher if mode == :async
end

Instance Attribute Details

#local_cursor ⇒ String?

How far the local database has been replayed.

Returns:

  • (String, nil) 


131
132
133
 
# File 'lib/discord_store/journal.rb', line 131

def local_cursor
  @local_cursor
end

#log ⇒ Object (readonly)

Returns the value of attribute log.



31
32
33
 
# File 'lib/discord_store/journal.rb', line 31

def log
  @log
end

#mode ⇒ Object (readonly)

Returns the value of attribute mode.



31
32
33
 
# File 'lib/discord_store/journal.rb', line 31

def mode
  @mode
end

Class Method Details

.build_configuration(options) ⇒ Configuration

Builds a Configuration from a database.yml stanza.

Parameters:

  • options (Hash)

Returns:



217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/discord_store/journal.rb', line 217

def self.build_configuration(options)
  Configuration.new.tap do |config|
    config.i_understand_this_violates_discord_tos =
      options[:i_understand_this_violates_discord_tos]
    config.token = options[:token]
    config.application_id = options[:application_id]
    config.guild_id = options[:guild_id]
    config.secret_key = options[:secret_key]
    config.cipher = (options[:cipher] || :aes_256_gcm).to_sym
    config.log_channel_ids = Array(options[:log_channel_ids]).map(&:to_s)
    config.logger = options[:logger]

    if config.log_channel_ids.size > 1
      raise ConfigurationError,
            "a write-ahead log needs one total order, and message IDs are only ordered " \
            "within a channel. Configure exactly one log_channel_id for the adapter, or " \
            "use DiscordStore::Log directly if per-stream ordering is enough."
    end
  end
end

.deserialize_binds(values) ⇒ Array

Parameters:

  • values (Array)

Returns:

  • (Array) 


196
197
198
 
# File 'lib/discord_store/journal.rb', line 196

def self.deserialize_binds(values)
  Array(values).map { |value| deserialize_value(value) }
end

.deserialize_value(value) ⇒ Object

Parameters:

  • value (Object)

Returns:

  • (Object) 


202
203
204
205
206
207
208
209
210
211
 
# File 'lib/discord_store/journal.rb', line 202

def self.deserialize_value(value)
  return value unless value.is_a?(Hash)

  if value.key?(BINARY_TAG) then Base64.strict_decode64(value[BINARY_TAG])
  elsif value.key?(TIME_TAG) then Time.iso8601(value[TIME_TAG])
  elsif value.key?(DATE_TAG) then Date.iso8601(value[DATE_TAG])
  elsif value.key?(DECIMAL_TAG) then BigDecimal(value[DECIMAL_TAG])
  else value
  end
end

.serialize_binds(binds) ⇒ Array

Turns ActiveRecord bind parameters into something JSON can carry, with tags for the types it cannot.

Parameters:

  • binds (Array)

Returns:

  • (Array) 


161
162
163
164
165
166
 
# File 'lib/discord_store/journal.rb', line 161

def self.serialize_binds(binds)
  Array(binds).map do |bind|
    value = bind.respond_to?(:value_for_database) ? bind.value_for_database : bind
    serialize_value(value)
  end
end

.serialize_value(value) ⇒ Object

Parameters:

  • value (Object)

Returns:

  • (Object) 


170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/discord_store/journal.rb', line 170

def self.serialize_value(value)
  case value
  when nil, true, false, Integer, Float then value
  when BigDecimal then { DECIMAL_TAG => value.to_s("F") }
  when Time then { TIME_TAG => value.utc.iso8601(6) }
  when DateTime then { TIME_TAG => value.to_time.utc.iso8601(6) }
  when Date then { DATE_TAG => value.iso8601 }
  when String
    # A UTF-8 string rides as itself; anything else is bytes, and bytes do
    # not survive JSON.
    if value.encoding == Encoding::BINARY || !value.valid_encoding?
      { BINARY_TAG => Base64.strict_encode64(value) }
    else
      value
    end
  else
    if value.respond_to?(:to_s) && value.class.name.to_s.include?("Binary")
      { BINARY_TAG => Base64.strict_encode64(value.to_s) }
    else
      value.to_s
    end
  end
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.



148
149
150
151
152
 
# File 'lib/discord_store/journal.rb', line 148

def close
  flush!
  @flusher&.kill
  @client&.close
end

#each_transaction(from: nil) {|entry, cursor| ... } ⇒ void

This method returns an undefined value.

Reads transactions in order.

Parameters:

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

    exclusive cursor

Yield Parameters:

  • entry (Hash) —

    =>, "at" =>

  • cursor (String) 


114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/discord_store/journal.rb', line 114

def each_transaction(from: nil)
  # Deliberately checks for a log rather than for recording?: a replay reads
  # while paused, and gating this on the write mode would make it a no-op
  # exactly when it matters.
  return if @log.nil?

  @log.each(stream: STREAM, after: from) do |record|
    data = record.data
    next unless data.is_a?(Hash) && data["tx"]

    yield data, record.cursor
  end
end

#flush! ⇒ Integer

Writes anything buffered.

Returns:

  • (Integer) —

    entries written



84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/discord_store/journal.rb', line 84

def flush!
  return 0 unless @mode == :async

  pending = @mutex.synchronize do
    drained = @buffer
    @buffer = []
    drained
  end
  return 0 if pending.empty?

  append(pending)
  pending.size
end

#pending_count ⇒ Integer

Returns entries waiting to be written.

Returns:

  • (Integer) —

    entries waiting to be written



99
 
# File 'lib/discord_store/journal.rb', line 99

def pending_count = @mutex.synchronize { @buffer.size }

#recording? ⇒ Boolean

Returns:

  • (Boolean) 


56
 
# File 'lib/discord_store/journal.rb', line 56

def recording? = @mode != :off

#tip ⇒ String?

Returns cursor of the newest record in the channel.

Returns:

  • (String, nil) —

    cursor of the newest record in the channel



102
103
104
105
106
 
# File 'lib/discord_store/journal.rb', line 102

def tip
  return nil unless recording?

  @log.tip(stream: STREAM)
end

#while_paused ⇒ Object

Suspends writing for the duration of the block, leaving reads working.

Replay uses this: applying a statement locally must not journal it back, or every replay would duplicate the log it just read.

Returns:

  • (Object) —

    the block's value



139
140
141
142
143
144
145
 
# File 'lib/discord_store/journal.rb', line 139

def while_paused
  previous = @mode
  @mode = :off
  yield
ensure
  @mode = previous
end

#write(statements) ⇒ String?

Records one transaction.

Parameters:

  • statements (Array<Hash>) —

    each =>, "binds" =>

Returns:

  • (String, nil) —

    the cursor the transaction landed at, if written now



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/discord_store/journal.rb', line 62

def write(statements)
  return nil unless recording?
  return nil if statements.empty?

  entry = { "tx" => statements, "at" => Time.now.utc.iso8601(3) }

  case @mode
  when :sync
    append([entry]).last&.cursor
  when :async
    buffered = @mutex.synchronize do
      @buffer << entry
      @buffer.size
    end
    flush! if buffered >= @max_buffer
    nil
  end
end