Class: RcrewAI::Rails::Observation::Writer

Inherits:
Object
  • Object
show all
Defined in:
lib/rcrewai/rails/observation/writer.rb

Overview

Persists spans, isolating the crew run from any storage failure.

Span creates always write through, because the collector needs the id to nest children. Only span events are buffered in :batched mode.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(mode: :batched, flush_every: 25, logger: nil, on_span_change: nil) ⇒ Writer

Returns a new instance of Writer.



13
14
15
16
17
18
19
20
21
 
# File 'lib/rcrewai/rails/observation/writer.rb', line 13

def initialize(mode: :batched, flush_every: 25, logger: nil, on_span_change: nil)
  @mode = mode
  @flush_every = flush_every
  @logger = logger
  @on_span_change = on_span_change
  @buffer = []
  @mutex = Mutex.new
  @dropped_count = 0
end

Instance Attribute Details

#dropped_count ⇒ Object (readonly)

Returns the value of attribute dropped_count.



11
12
13
 
# File 'lib/rcrewai/rails/observation/writer.rb', line 11

def dropped_count
  @dropped_count
end

Instance Method Details

#create_event(attrs) ⇒ Object

Span events carry no id that anything else references, so in :batched mode they buffer and insert in bulk.



43
44
45
46
47
48
49
50
51
52
 
# File 'lib/rcrewai/rails/observation/writer.rb', line 43

def create_event(attrs)
  return guard { SpanEvent.create!(attrs).id } if @mode == :immediate

  should_flush = @mutex.synchronize do
    @buffer << attrs.merge(created_at: Time.current, updated_at: Time.current)
    @buffer.size >= @flush_every
  end
  flush! if should_flush
  nil
end

#create_span(attrs) ⇒ Object

Returns the span id, or nil if the write failed.



24
25
26
27
28
29
 
# File 'lib/rcrewai/rails/observation/writer.rb', line 24

def create_span(attrs)
  guard do
    span = Span.create!(attrs)
    span.id
  end.tap { |id| notify_span_change(id) if id }
end

#flush! ⇒ Object

Flushes buffered span events. Span creates are never buffered (see create_span), so this only drains the event buffer.



56
57
58
59
60
61
 
# File 'lib/rcrewai/rails/observation/writer.rb', line 56

def flush!
  buffered = @mutex.synchronize { @buffer.slice!(0..-1) || [] }
  return if buffered.empty?

  guard { SpanEvent.insert_all(buffered) }
end

#update_span(span_id, attrs) ⇒ Object

Writes through update_all, which fires no ActiveRecord callbacks — so span completion would be invisible to any after_update hook. The change notification is therefore raised explicitly here.



34
35
36
37
38
39
 
# File 'lib/rcrewai/rails/observation/writer.rb', line 34

def update_span(span_id, attrs)
  guard do
    Span.where(id: span_id).update_all(attrs.merge(updated_at: Time.current))
    span_id
  end.tap { |id| notify_span_change(id) if id }
end