Class: Dis::Storage

Inherits:
Object
  • Object
show all
Defined in:
lib/dis/storage.rb

Overview

Dis Storage

Interface for interacting with the storage layers.

All queries are scoped by object type, which will default to the table name of the model. Take care to use your own scope if you interact with the store directly, as models will purge expired content when they change.

Files are stored with a SHA1 digest of the file contents as the key. This ensures data is deduplicated per scope. Hash collisions will be silently ignored.

Layers should be added to Dis::Storage.layers. At least one writeable, non-delayed layer must exist.

Class Method Summary collapse

Class Method Details

.change_type(prev_type, new_type, key) ⇒ String

Changes the type of an object. Kicks off a Jobs::ChangeType job if any delayed layers are defined.

Examples:

Dis::Storage.change_type("old_things", "new_things", key)

Parameters:

  • prev_type (String) —

    the current type scope

  • new_type (String) —

    the new type scope

  • key (String) —

    the content hash

Returns:

  • (String) —

    the content hash

Raises:



60
61
62
63
64
65
66
67
68
69
 
# File 'lib/dis/storage.rb', line 60

def change_type(prev_type, new_type, key)
  require_writeable_layers!
  file = get(prev_type, key)
  store_immediately!(new_type, file)
  layers.immediate.writeable.each do |layer|
    layer.delete(prev_type, key)
  end
  enqueue_delayed_jobs(prev_type, new_type, key)
  key
end

.delayed_delete(type, key) ⇒ void

This method returns an undefined value.

Deletes content from all delayed layers. Called internally by Jobs::Delete.

Parameters:

  • type (String) —

    the type scope

  • key (String) —

    the content hash



272
273
274
275
276
 
# File 'lib/dis/storage.rb', line 272

def delayed_delete(type, key)
  layers.delayed.writeable.each do |layer|
    layer.delete(type, key)
  end
end

.delayed_store(type, hash) ⇒ void

This method returns an undefined value.

Transfers files from immediate layers to all delayed layers. Called internally by Jobs::Store.

Parameters:

  • type (String) —

    the type scope

  • hash (String) —

    the content hash

Raises:



98
99
100
101
102
103
 
# File 'lib/dis/storage.rb', line 98

def delayed_store(type, hash)
  file = get(type, hash)
  layers.delayed.writeable.each do |layer|
    layer.store(type, hash, file)
  end
end

.delete(type, key) ⇒ Boolean

Deletes a file from all layers. Kicks off a Jobs::Delete job if any delayed layers are defined.

Examples:

Dis::Storage.delete("things", key) # => true
Dis::Storage.delete("things", key) # => false

Parameters:

  • type (String) —

    the type scope

  • key (String) —

    the content hash

Returns:

  • (Boolean) —

    true if the file existed in any immediate layer

Raises:



205
206
207
208
209
210
211
212
213
 
# File 'lib/dis/storage.rb', line 205

def delete(type, key)
  require_writeable_layers!
  deleted = false
  layers.immediate.writeable.each do |layer|
    deleted = true if layer.delete(type, key)
  end
  Dis::Jobs::Delete.perform_later(type, key) if layers.delayed.writeable.any?
  deleted
end

.evict_caches ⇒ void

This method returns an undefined value.

Evicts cached files from all cache layers that exceed their size limit. Only evicts files that have been replicated to a non-cache writeable layer.



220
221
222
 
# File 'lib/dis/storage.rb', line 220

def evict_caches
  layers.cache.each { |layer| evict_cache(layer) }
end

.exists?(type, key) ⇒ Boolean

Returns true if the file exists in any layer.

Examples:

Dis::Storage.exists?("things", key) # => true

Parameters:

  • type (String) —

    the type scope

  • key (String) —

    the content hash

Returns:

  • (Boolean)

Raises:



114
115
116
117
118
119
120
121
122
 
# File 'lib/dis/storage.rb', line 114

def exists?(type, key)
  require_layers!
  layers.each do |layer|
    return true if layer.exists?(type, key)
  rescue StandardError => e
    report_layer_error(e, layer:, type:, key:)
  end
  false
end

.file_digest(file) {|hash| ... } ⇒ String

Returns a hex digest for a given binary. Accepts File/IO objects, strings, and Fog models.

Parameters:

  • file (File, IO, String, Fog::Model) —

    the content to digest

Yields:

  • (hash) —

    if a block is given, yields the hex digest

Yield Parameters:

  • hash (String) —

    the computed SHA1 hex digest

Returns:

  • (String) —

    the SHA1 hex digest



27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/dis/storage.rb', line 27

def file_digest(file)
  hash = case file
         when Fog::Model
           digest.hexdigest(file.body)
         when String
           digest.hexdigest(file)
         else
           digest.file(file.path).hexdigest
         end
  yield hash if block_given?
  hash
end

.file_path(type, key) ⇒ String?

Returns the absolute file path from the first layer that has a local copy, or nil if no layer stores files locally.

Parameters:

  • type (String) —

    the type scope

  • key (String) —

    the content hash

Returns:

  • (String, nil) —

    the absolute file path, or nil

Raises:



158
159
160
161
162
163
164
165
166
167
 
# File 'lib/dis/storage.rb', line 158

def file_path(type, key)
  require_layers!
  layers.each do |layer|
    path = layer.file_path(type, key)
    return path if path
  rescue StandardError => e
    report_layer_error(e, layer:, type:, key:)
  end
  nil
end

.get(type, key) ⇒ Fog::Model

Retrieves a file from the store. If the first layer misses, the file is fetched from the next available layer and backfilled to all immediate layers.

Examples:

file = Dis::Storage.get("things", hash)
file.body # => "file contents..."

Parameters:

  • type (String) —

    the type scope

  • key (String) —

    the content hash

Returns:

  • (Fog::Model) —

    the stored file

Raises:



138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/dis/storage.rb', line 138

def get(type, key)
  require_layers!
  fetch_count = 0
  result = layers.inject(nil) do |res, layer|
    next res if res

    fetch_count += 1
    fetch_from_layer(layer, type, key)
  end || raise(Dis::Errors::NotFoundError)
  backfill!(type, result) if fetch_count > 1
  result
end

.get_file(type, key, file) ⇒ File

Streams the contents of a file into the given file, fetching from the first layer that has it. Backfills faster layers if the content had to be fetched from further down.

Parameters:

  • type (String) —

    the type scope

  • key (String) —

    the content hash

  • file (File) —

    the destination, must respond to path

Returns:

  • (File) —

    the file that was written to, positioned at the start

Raises:



180
181
182
183
184
185
186
187
188
189
190
 
# File 'lib/dis/storage.rb', line 180

def get_file(type, key, file)
  require_layers!
  fetch_count = 0
  found = layers.detect do |layer|
    fetch_count += 1
    stream_from_layer(layer, type, key, file)
  end
  raise Dis::Errors::NotFoundError unless found

  finalize_fetched_file(type, file, fetch_count)
end

.layers ⇒ Dis::Layers

Exposes the layer set.

Returns:



43
44
45
 
# File 'lib/dis/storage.rb', line 43

def layers
  @layers ||= Dis::Layers.new
end

.missing_keys(model) {|batch_size| ... } ⇒ Array<String>

Returns content hashes from the model's table that exist in no non-cache layer.

Examples:

Dis::Storage.missing_keys(Image)

Parameters:

  • model (Class) —

    an ActiveRecord model that includes Model

Yields:

  • (batch_size) —

    called after each batch is checked

Yield Parameters:

  • batch_size (Integer) —

    the number of keys in the batch

Returns:

  • (Array<String>) —

    content hashes with no backing file



236
237
238
239
240
241
242
243
244
245
246
 
# File 'lib/dis/storage.rb', line 236

def missing_keys(model)
  attr = model.dis_attributes[:content_hash]
  missing = []

  model.where.not(attr => nil).in_batches(of: 200) do |batch|
    keys = batch.pluck(attr)
    missing.concat(uncovered_keys(keys.uniq, model.dis_type))
    yield keys.size if block_given?
  end
  missing.uniq
end

.orphaned_keys(model) ⇒ Hash{Dis::Layer => Array<String>}

Returns a hash of layer => orphaned content hashes for files that exist in storage but have no matching database record.

Examples:

Dis::Storage.orphaned_keys(Image)

Parameters:

  • model (Class) —

    an ActiveRecord model that includes Model

Returns:

  • (Hash{Dis::Layer => Array<String>}) —

    orphaned content hashes per layer



258
259
260
261
262
263
264
 
# File 'lib/dis/storage.rb', line 258

def orphaned_keys(model)
  layers.non_cache.each_with_object({}) do |layer, result|
    orphans = layer_orphans(layer, model.dis_type, model,
                            model.dis_attributes[:content_hash])
    result[layer] = orphans if orphans.any?
  end
end

.store(type, file) ⇒ String

Stores a file and returns a content hash. Kicks off a Jobs::Store job if any delayed layers are defined.

Examples:

hash = Dis::Storage.store("things", File.open("foo.bin"))
# => "8843d7f92416211de9ebb963ff4ce28125932878"

Parameters:

  • type (String) —

    the type scope (e.g. table name)

  • file (File, IO, String, Fog::Model) —

    the content to store

Returns:

  • (String) —

    the SHA1 content hash

Raises:



83
84
85
86
87
88
89
 
# File 'lib/dis/storage.rb', line 83

def store(type, file)
  require_writeable_layers!
  hash = store_immediately!(type, file)
  Dis::Jobs::Store.perform_later(type, hash) if layers.delayed.writeable.any?
  Dis::Jobs::Evict.perform_later if layers.cache?
  hash
end