Class: Onlylogs::Spool

Inherits:
Object
  • Object
show all
Defined in:
lib/onlylogs/spool.rb

Overview

A bounded, on-disk overflow buffer for log batches that could not be delivered.

HttpLogger keeps the happy path in memory: only when a send fails or the circuit is open does a batch get written here, to be replayed once the drain recovers (and on the next boot). This turns transient-failure / restart data loss into at-least-once delivery: a batch that was in fact received but whose response was lost will be replayed and show up as a duplicate downstream. Duplicates are an accepted trade for not losing data.

Constant Summary collapse

DEFAULT_MAX_BYTES =

128 MB

128 * 1024 * 1024

Instance Method Summary collapse

Constructor Details

#initialize(dir:, max_bytes: DEFAULT_MAX_BYTES) ⇒ Spool

Returns a new instance of Spool.



17
18
19
20
21
22
23
24
25
 
# File 'lib/onlylogs/spool.rb', line 17

def initialize(dir:, max_bytes: DEFAULT_MAX_BYTES)
  @dir = dir
  @max_bytes = max_bytes
  # Unique per instance so two runs (even with a reused pid) never collide on a filename.
  @token = SecureRandom.hex(4)
  @seq = 0
  @mutex = Mutex.new
  ::FileUtils.mkdir_p(@dir)
end

Instance Method Details

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


58
59
60
 
# File 'lib/onlylogs/spool.rb', line 58

def empty?
  pending_files.empty?
end

#replay ⇒ Object

Replay pending batches oldest-first. Yields each body; if the block returns truthy the file is deleted (delivered), otherwise replay stops and the remaining files are kept for later.



47
48
49
50
51
52
53
54
55
56
 
# File 'lib/onlylogs/spool.rb', line 47

def replay
  pending_files.each do |path|
    body = read(path)
    next if body.nil? # already claimed/deleted by another process

    break unless yield(body)

    delete(path)
  end
end

#write(body) ⇒ Object

Persist a batch body. Rolls the oldest batches off first if the byte cap would be exceeded.



28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/onlylogs/spool.rb', line 28

def write(body)
  return if body.nil? || body.empty?

  @mutex.synchronize do
    evict(body.bytesize)
    seq = (@seq += 1)
    final = ::File.join(@dir, "#{@token}-#{format("%09d", seq)}.batch")
    tmp = "#{final}.tmp"
    # Write to a temp name then rename: rename is atomic, so replay never reads a
    # half-written file (it only globs *.batch).
    ::File.binwrite(tmp, body)
    ::File.rename(tmp, final)
  end
rescue => e
  Kernel.warn "Onlylogs::Spool write error: #{e.class}: #{e.message}"
end