Class: Silas::Api::V1::StreamsController

Inherits:
BaseController
  • Object
show all
Includes:
ActionController::Live, Serialization
Defined in:
app/controllers/silas/api/v1/streams_controller.rb

Overview

Server-sent events at ROW granularity: turn / completed-step / invocation changes polled from the durable rows — the same source of truth as everything else. Deliberately NOT per-token: deltas are emitted in the worker process and the gem requires no cross-process bus; per-token streaming is the Turbo/browser feature.

Delivery is at-least-once with Last-Event-ID resume: event ids are epoch-millisecond watermarks, each poll re-reads a 1ms overlap, and clients dedup by (event type, id, status). ?poll=1 emits the backlog since Last-Event-ID and closes — curl-able, and how the request specs exercise the cursor without holding a live stream.

Live actions run on their own thread and hold a Puma thread for the stream's lifetime — inherent to SSE. The AR connection is NOT held: each poll checks one out and returns it before sleeping, and the stream closes itself after api_stream_max_duration (clients reconnect with Last-Event-ID).

Constant Summary collapse

HEARTBEAT_EVERY =

seconds; comment-frames keep proxies alive

15

Instance Method Summary collapse

Methods included from Serialization

#invocation_json, #session_json, #step_json, #turn_json

Instance Method Details

#show ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'app/controllers/silas/api/v1/streams_controller.rb', line 29

def show
  session_id = with_connection { Silas::Session.find(params[:id]).id }

  response.headers["Content-Type"] = "text/event-stream"
  response.headers["Cache-Control"] = "no-cache"
  response.headers["X-Accel-Buffering"] = "no" # nginx: never buffer SSE
  # Whatever the router/auth checked out must not stay pinned for the
  # stream's lifetime.
  ActiveRecord::Base.connection_handler.clear_active_connections!

  watermark = initial_watermark
  deadline = clock + Silas.config.api_stream_max_duration
  last_beat = clock

  loop do
    events = with_connection { collect_changes(session_id, watermark) }
    events.each do |type, at, payload|
      write_event(type, payload, watermark_ms(at))
      watermark = at if at > watermark
    end

    break if params[:poll].present? # backlog served; close

    if clock >= deadline
      write_event("timeout", { reconnect: true }, watermark_ms(watermark))
      break
    end

    if clock - last_beat >= HEARTBEAT_EVERY
      response.stream.write(": hb\n\n")
      last_beat = clock
    end

    sleep Silas.config.api_stream_poll_interval
  end
rescue IOError, ActionController::Live::ClientDisconnected
  # client went away — the normal end of an SSE stream
ensure
  response.stream.close rescue nil
end