Class: DiscordStore::Log

Inherits:
Object
  • Object
show all
Defined in:
lib/discord_store/log.rb,
lib/discord_store/log/record.rb

Overview

An append-only log whose storage is a set of Discord channels.

This is the layer everything else is built on, and it is append-only for a reason that is worth stating plainly, because it is the joke at the centre of this library.

Deleting from Discord is expensive and gets worse with age: bulk deletion only covers messages under two weeks old, and past that it is one request per message against a rate limit that is already the binding constraint. So the cheap way to retire a record is to append a marker saying it is gone and let the reader skip it.

That marker is a tombstone. Discord's own storage layer, Cassandra, did the same thing for the same reason — and the volume of tombstones a chat app generates is precisely what made Cassandra untenable for them and forced the migration to ScyllaDB. Building a store on Discord means writing tombstones into a database whose tombstones are stored as tombstones.

Compaction is therefore not optional at scale; see #compact.

Defined Under Namespace

Classes: Buffer, Record

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(rest:, config:, channel_ids: nil, cipher: nil) ⇒ Log

Returns a new instance of Log.

Parameters:

  • rest (Transport::REST)
  • config (Configuration)
  • channel_ids (Array<String>, nil) (defaults to: nil) —

    defaults to config.log_channel_ids

  • cipher (#seal, #open, nil) (defaults to: nil) 


36
37
38
39
40
41
42
 
# File 'lib/discord_store/log.rb', line 36

def initialize(rest:, config:, channel_ids: nil, cipher: nil)
  @rest = rest
  @config = config
  @shard = ChannelShard.new(channel_ids || config.log_channel_ids)
  @cipher = cipher || Cipher.build(config)
  @codec = Codec.new(cipher: @cipher, content_budget: config.content_budget)
end

Instance Attribute Details

#rest ⇒ Transport::REST (readonly)

Returns:



30
31
32
 
# File 'lib/discord_store/log.rb', line 30

def rest
  @rest
end

#shard ⇒ ChannelShard (readonly)

Returns:



27
28
29
 
# File 'lib/discord_store/log.rb', line 27

def shard
  @shard
end

Instance Method Details

#append(stream:, data:, nonce: nil) ⇒ Log::Record

Appends one record.

Parameters:

  • stream (String) —

    logical topic and partition key

  • data (Hash)
  • nonce (String, nil) (defaults to: nil) —

    pass a stable value to make a retry idempotent

Returns:



50
51
52
 
# File 'lib/discord_store/log.rb', line 50

def append(stream:, data:, nonce: nil)
  append_all([Record.new(stream: stream, data: data, nonce: nonce)]).first
end

#append_all(records, atomic: false) ⇒ Array<Log::Record>

Appends several records.

Records are grouped by partition and packed, so N records cost far fewer than N requests. With atomic, every record must land in one message or the call raises: one message is the largest unit Discord commits all-or-nothing, and that is the only atomicity on offer here.

Parameters:

  • records (Array<Log::Record>)
  • atomic (Boolean) (defaults to: false)

Returns:

  • (Array<Log::Record>) —

    the same records, with lsn assigned



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

def append_all(records, atomic: false)
  records = Array(records)
  return [] if records.empty?

  by_channel = records.group_by { |record| @shard.for(record.stream) }

  if atomic && by_channel.size > 1
    raise PayloadTooLargeError,
          "an atomic batch spans #{by_channel.size} partitions (#{by_channel.keys.join(", ")}). " \
          "Records that must commit together must share a partition; give them the same " \
          "stream, or configure a single log channel."
  end

  by_channel.flat_map do |channel_id, channel_records|
    write_to_channel(channel_id, channel_records, atomic: atomic)
  end
end

#compact(stream:, dry_run: false) ⇒ Hash

Discards tombstoned records by physically deleting their messages.

This is the expensive path, and the only one that reclaims anything. It is deliberately explicit rather than automatic: it costs one request per message once the two-week bulk-delete window has closed, which at the per-channel limit is roughly one message per second.

Parameters:

  • stream (String)
  • dry_run (Boolean) (defaults to: false) —

    report what would be deleted without deleting

Returns:

  • (Hash) —

    tombstoned:, deleted:, too_old:



187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/discord_store/log.rb', line 187

def compact(stream:, dry_run: false)
  channel_id = @shard.for(stream)
  targets = []
  scanned = 0

  read_channel(channel_id, stream: stream) do |record|
    scanned += 1
    targets << record.data["target"] if record.tombstone?
  end

  doomed = targets.filter_map { |cursor| cursor.to_s.split(":").first }.uniq

  return { scanned: scanned, tombstoned: doomed.size, deleted: 0, too_old: 0, dry_run: true } if dry_run

  too_old = @rest.bulk_delete_messages(channel_id, doomed)

  if @config.delete_policy == :aggressive
    too_old.each { |id| @rest.delete_message(channel_id, id) }
    too_old = []
  end

  { scanned: scanned, tombstoned: doomed.size,
    deleted: doomed.size - too_old.size, too_old: too_old.size, dry_run: false }
end

#each(stream: nil, after: nil, limit: nil) {|record| ... } ⇒ Array<Log::Record>, void

Reads records in partition order.

Parameters:

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

    a single stream, or nil for every partition

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

    a cursor from DiscordStore::Log::Record#cursor; exclusive

  • limit (Integer, nil) (defaults to: nil)

Yield Parameters:

Returns:



108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/discord_store/log.rb', line 108

def each(stream: nil, after: nil, limit: nil, &block)
  return enum_for(:each, stream: stream, after: after, limit: limit) unless block

  channels = stream ? [@shard.for(stream)] : @shard.channel_ids
  count = 0

  channels.each do |channel_id|
    read_channel(channel_id, after: after, stream: stream) do |record|
      block.call(record)
      count += 1
      # Non-local on purpose: the limit applies to the whole walk across
      # every partition, not to the current channel.
      return if limit && count >= limit # rubocop:disable Lint/NonLocalExitFromIterator
    end
  end
end

#read(stream: nil, after: nil, limit: nil) ⇒ Array<Log::Record>

Returns:



126
127
128
 
# File 'lib/discord_store/log.rb', line 126

def read(stream: nil, after: nil, limit: nil)
  each(stream: stream, after: after, limit: limit).to_a
end

#replace(record, data:) ⇒ Log::Record

Rewrites a message in place. Only valid for a message holding exactly one record; a packed message cannot be edited record-wise without rewriting its neighbours, and rewriting history in an append-only log is how replays start disagreeing with each other.

Parameters:

Returns:

Raises:

  • (ArgumentError) 


160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/discord_store/log.rb', line 160

def replace(record, data:)
  raise ArgumentError, "record has no lsn" unless record.lsn

  channel_id = record.channel_id || @shard.for(record.stream)
  updated = Record.new(stream: record.stream, data: data, nonce: record.nonce,
                       lsn: record.lsn, position: 0, channel_id: channel_id)

  packed = @codec.pack([updated], aad: channel_id, atomic: true).first

  unless packed[:files].empty?
    raise PayloadTooLargeError, "replacement spills to an attachment; append a new record instead"
  end

  @rest.edit_message(channel_id, record.lsn, content: packed[:content])
  updated
end

#tip(stream: nil) ⇒ String?

The cursor of the most recent record in a partition, or nil if empty.

Parameters:

  • stream (String, nil) (defaults to: nil)

Returns:

  • (String, nil) 


134
135
136
137
138
139
140
141
 
# File 'lib/discord_store/log.rb', line 134

def tip(stream: nil)
  channel_id = stream ? @shard.for(stream) : @shard.channel_ids.first
  message = @rest.list_messages(channel_id, limit: 1).first
  return nil unless message && @codec.ours?(message)

  records = decode(message)
  records.last&.cursor
end

#tombstone(stream:, target:) ⇒ Log::Record

Appends a tombstone for target.

Parameters:

  • stream (String)
  • target (String) —

    the cursor or key being retired

Returns:



148
149
150
 
# File 'lib/discord_store/log.rb', line 148

def tombstone(stream:, target:)
  append_all([Record.tombstone(stream: stream, target: target)]).first
end

#transaction {|buffer| ... } ⇒ Array<Log::Record>

Buffers appends and writes them as one atomic batch on success.

log.transaction do |tx|
tx.append(stream: "orders", data: { id: 1 })
tx.append(stream: "orders", data: { id: 2 })
end

Nothing is written if the block raises.

Yield Parameters:

Returns:



93
94
95
96
97
98
99
 
# File 'lib/discord_store/log.rb', line 93

def transaction
  buffer = Buffer.new
  yield buffer
  return [] if buffer.empty?

  append_all(buffer.records, atomic: true)
end