Class: DiscordStore::Codec

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

Overview

Turns records into message payloads and back.

A Discord message holds 2000 characters, and a request that carries one 80-byte record is a request that spent the same rate-limit permit as one carrying twenty. Since permits are the scarce resource, packing is not an optimisation here — it is most of the throughput.

Wire format of a message:

DS1 i 3            <- header: version, kind, record count
<envelope>         <- one record per line, base64url so a line break
<envelope>            can never appear inside one
<envelope>

A record too large to fit alongside anything else spills to an attachment and the message body becomes a stub:

DS1 s 1

Bigger than an attachment is the blob store's problem, not the log's.

Defined Under Namespace

Classes: Batch

Constant Summary collapse

HEADER_PREFIX = 
"DS1"
KIND_INLINE = 
"i"
KIND_SPILL = 
"s"
SPILL_FILENAME = 
"records.ds1"
SEPARATOR = 
"\n"

Instance Method Summary collapse

Constructor Details

#initialize(cipher:, content_budget: 1900, spill_limit: nil) ⇒ Codec

Returns a new instance of Codec.

Parameters:

  • cipher (#seal, #open)
  • content_budget (Integer) (defaults to: 1900) —

    characters we are willing to use

  • spill_limit (Integer, nil) (defaults to: nil) —

    max attachment bytes; nil means unlimited



37
38
39
40
41
 
# File 'lib/discord_store/codec.rb', line 37

def initialize(cipher:, content_budget: 1900, spill_limit: nil)
  @cipher = cipher
  @content_budget = content_budget
  @spill_limit = spill_limit
end

Instance Method Details

#ours?(message) ⇒ Boolean

Whether a message looks like something this library wrote. Cheap enough to run before doing any crypto.

Parameters:

  • message (Hash)

Returns:

  • (Boolean) 


123
124
125
 
# File 'lib/discord_store/codec.rb', line 123

def ours?(message)
  message.is_a?(Hash) && message["content"].to_s.start_with?("#{HEADER_PREFIX} ")
end

#pack(records, aad: nil, atomic: false) ⇒ Array<Hash>

Packs records into as few messages as will hold them.

Parameters:

  • records (Array<Log::Record>)
  • aad (String, nil) (defaults to: nil) —

    bound into each envelope's authentication tag

  • atomic (Boolean) (defaults to: false) —

    when true, refuse to split the batch across messages: either it fits in one message or it raises. One message is one atomic append, so this is how a transaction gets all-or-nothing durability on a platform with no transactions.

Returns:

  • (Array<Hash>) —

    each files:, records:

Raises:



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
 
# File 'lib/discord_store/codec.rb', line 53

def pack(records, aad: nil, atomic: false)
  messages = []
  batch = Batch.new(inline_budget)

  records.each do |record|
    envelope = seal(record, aad)

    if envelope.length > inline_budget
      raise_if_atomic(atomic, "a single record exceeds one message")
      messages << current_message(batch.drain) if batch.any?
      messages << spill_message(record, envelope)
      next
    end

    unless batch.fits?(envelope)
      raise_if_atomic(atomic, "the batch does not fit in one message")
      messages << current_message(batch.drain)
    end

    batch.add(record, envelope)
  end

  messages << current_message(batch.drain) if batch.any?
  messages
end

#spilled?(message) ⇒ Boolean

Returns whether the payload lives in an attachment.

Parameters:

  • message (Hash)

Returns:

  • (Boolean) —

    whether the payload lives in an attachment



129
130
131
132
 
# File 'lib/discord_store/codec.rb', line 129

def spilled?(message)
  header, = split_header(message["content"].to_s)
  header && header[:kind] == KIND_SPILL
end

#unpack(message, attachment_body: nil, aad: nil) ⇒ Array<Log::Record>

Reverses #pack for one Discord message.

Parameters:

  • message (Hash) —

    a Discord message object

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

    the spilled bytes, if the message has a spill attachment the caller has already fetched

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

Returns:



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
 
# File 'lib/discord_store/codec.rb', line 86

def unpack(message, attachment_body: nil, aad: nil)
  content = message["content"].to_s
  header, body = split_header(content)
  return [] if header.nil?

  raw = if header[:kind] == KIND_SPILL
          unless attachment_body
            raise CorruptRecordError,
                  "message #{message["id"]} spilled to an attachment that was not fetched"
          end

          attachment_body
        else
          body
        end

  lines = raw.split(SEPARATOR).reject(&:empty?)

  if header[:count] != lines.size
    raise CorruptRecordError,
          "message #{message["id"]} claims #{header[:count]} records but carries #{lines.size}"
  end

  lines.each_with_index.map do |envelope, position|
    payload = JSON.parse(@cipher.open(envelope, aad: aad))
    Log::Record.from_wire(payload, lsn: message["id"], position: position,
                                   channel_id: message["channel_id"])
  end
rescue JSON::ParserError => e
  raise CorruptRecordError, "message #{message["id"]} holds invalid JSON: #{e.message}"
end